package cn.getech.data.development.service.impl;

import cn.getech.data.development.config.properties.BdpJobConfig;
import cn.getech.data.development.config.properties.DataDevelopmentConfig;
import cn.getech.data.development.config.properties.OozieConfig;
import cn.getech.data.development.constant.*;
import cn.getech.data.development.dto.*;
import cn.getech.data.development.entity.*;
import cn.getech.data.development.feign.JobRestFeignApi;
import cn.getech.data.development.mapper.*;
import cn.getech.data.development.model.dto.LogAndIdDto;
import cn.getech.data.development.model.dto.WorkFlowDepDto;
import cn.getech.data.development.model.dto.WorkMenuDto;
import cn.getech.data.development.model.qo.DevOpsQO;
import cn.getech.data.development.model.res.procinfo.ProcInfoListRes;
import cn.getech.data.development.model.vo.ProcWithWorkMenusVO;
import cn.getech.data.development.model.vo.WorkFlowStatusVO;
import cn.getech.data.development.model.vo.WorkSqlVO;
import cn.getech.data.development.service.*;
import cn.getech.data.development.utils.DataPermissionUtil;
import cn.getech.data.development.utils.HdfsUtil;
import cn.getech.data.development.utils.TimeUtil;
import cn.getech.data.development.utils.hadoop.HadoopLogUtils;
import cn.getech.data.development.utils.hadoop.KillSparkTaskUtils;
import cn.getech.data.development.utils.hadoop.ShellSparkUtils;
import cn.getech.data.development.utils.oozie.JobConfUtil;
import cn.getech.data.development.utils.oozie.JobUtil;
import cn.getech.data.development.utils.ws.WorkFlowLayStatusSocketServer;
import cn.getech.data.development.utils.ws.WorkFlowStatusSocketServer;
import cn.getech.data.intelligence.common.exception.RRException;
import cn.getech.data.intelligence.common.utils.*;
import cn.getech.system.center.entity.SysUserEntity;
import cn.getech.system.center.mapper.SysUserMapper;
import cn.getech.system.center.service.SysRoleService;
import cn.getech.system.center.service.SysUserRoleService;
import cn.getech.system.center.utils.ShiroUtils;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateTime;
import cn.hutool.core.thread.ThreadUtil;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.ObjectUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.reflect.FieldUtils;
import org.apache.oozie.client.OozieClient;
import org.apache.oozie.client.OozieClientException;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.websocket.Session;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

/**
 * <p>
 * 作业流菜单树 服务实现类
 * </p>
 *
 * @author hujz
 * @since 2020-04-13
 */
@Slf4j
@Service
public class WorkMenuServiceImpl extends ServiceImpl<WorkMenuMapper, WorkMenu> implements IWorkMenuService {

    @Autowired
    private WorkMenuMapper workMenuMapper;

    @Autowired
    private WorkMenuJobRelMapper workMenuJobRelMapper;
    @Autowired
    private JobNodeInfoMapper jobNodeInfoMapper;
    @Autowired
    private JobInfoMapper jobInfoMapper;
    @Autowired
    private ProcInfoMapper procInfoMapper;
    @Autowired
    private WorkFlowDepMapper workFlowDepMapper;
    @Autowired
    private JobLinkMapper jobLinkMapper;
    @Autowired
    private JobNodeLayConfigMapper jobNodeLayConfigMapper;
    @Autowired
    private JobNodeConfMapper jobNodeConfMapper;
    @Autowired
    private ProcUserMapper procUserMapper;

    @Autowired
    private JobNodeInfoService jobNodeInfoService;

    @Autowired
    private OozieConfig oozieConfig;

    @Autowired
    private BdpJobConfig bdpJobConfig;

    @Autowired
    private JobInfoService jobInfoService;

    @Autowired
    private JobRunHistoryService jobRunHistoryService;

    @Autowired
    private SysUserMapper sysUserMapper;

    @Autowired
    private WorkFlowDepService workFlowDepService;

    @Autowired
    private JobDepRunHistoryMapper jobDepRunHistoryMapper;
    @Autowired
    private SysUserRoleService sysUserRoleService;
    @Autowired
    private JobNodeRunHistoryService jobNodeRunHistoryService;
    @Autowired
    private IWorkMenuService iWorkMenuService;
    @Autowired
    private IBdpVersionService iBdpVersionService;
    @Autowired
    private RealtimeQueueMapper realtimeQueueMapper;
    @Autowired
    private BussessConfigMapper bussessConfigMapper;

    @Autowired
    private JobRestFeignApi jobRestFeignApi;
    @Autowired
    private JobNodeConfService jobNodeConfService;

    @Autowired
    private BdpModelTaskMapper bdpModelTaskMapper;

    @Autowired
    private SysRoleService sysRoleService;
    @Autowired
    private RealTimeTaskMapper realTimeTaskMapper;
    @Autowired
    private ScheduleTimeService scheduleTimeService;
    @Autowired
    private DataDevelopmentConfig dataDevelopmentConfig;

    @Override
    public WorkMenu addsubWorkFlowMenu(WorkMenuDto workMenuDto) {
        //验证参数
        if (null == workMenuDto) {
            throw new RRException(DataDevelopmentBizExceptionEnum.PARAM_ERROR.getMessage());
        }
        //验证需要新增的子目录
        if (null == workMenuDto.getParentId() || 0L == workMenuDto.getParentId()) {
            throw new RRException("不能新增根目录!");
        }
        //名字必须传
        if (StringUtils.isEmpty(workMenuDto.getName())) {
            throw new RRException("目录名字必须传递!");
        }
        if (!Objects.equals(0L, workMenuDto.getParentId())) {
            WorkMenu workMenu = workMenuMapper.selectById(workMenuDto.getParentId());
            if (null == workMenu) {
                throw new RRException("父级工作流不存在!");
            }
            //验证是否可以添加目录
            if (!Objects.equals(workMenu.getWorkType(), WorkFlowMenuType.ROOT_DIR.getCode())
                    && !Objects.equals(workMenu.getWorkType(), WorkFlowMenuType.WORKFLOW_MGT.getCode())
                    && !Objects.equals(workMenu.getWorkType(), WorkFlowMenuType.SUB_DIR.getCode())
                    && !Objects.equals(workMenu.getWorkType(), WorkFlowMenuType.SUB_SUB_DIR.getCode())) {
                throw new RRException("父级菜单不能添加目录!");
            }
            workMenuDto.setRootId(workMenu.getRootId());
        }

        if (null == workMenuDto.getProcId()) {
            throw new RRException("工程id不能为空!");
        }
        //验证对应的用户是否可以操作这个工程
        cheackProcId(workMenuDto.getProcId(), ShiroUtils.getUserId(), false, null,null,null);

        //验证当前工程中的名字
        cheackProcIdAndName(workMenuDto.getProcId(), workMenuDto.getParentId(), workMenuDto.getName(), null);

        //新增目录
        return saveMenu(workMenuDto);

    }

    private void cheackProcIdAndName(Integer procId, Long parentId, String name, Long workMenuId) {
        List<WorkMenu> workMenus = workMenuMapper.selectList(new QueryWrapper<WorkMenu>()
                .eq("proc_id", procId)
                .eq("name", name)
                .eq(null != parentId,"parent_id", parentId)
                .ne(null != workMenuId, "id", workMenuId));
        if (CollectionUtil.isNotEmpty(workMenus)) {
            throw new RRException("相同的工程下面目录名字不能重复!");
        }
    }

    @Override
    public void delWorkFlowMenu(WorkMenuDto workMenuDto) throws Exception {
        //验证参数
        if (null == workMenuDto) {
            throw new RRException(DataDevelopmentBizExceptionEnum.PARAM_ERROR.getMessage());
        }
        if (null == workMenuDto.getProcId()) {
            throw new RRException("工程id不能为空!");
        }

        if (null == workMenuDto.getId()) {
            throw new RRException("目录菜单id不能为空");
        }

        WorkMenu workMenu = workMenuMapper.selectById(workMenuDto.getId());
        if (null == workMenu) {
            throw new RRException("目录菜单不存在");
        }
        if (workMenu.getWorkType().equals(WorkFlowMenuType.ROOT_DIR.getCode())
                ||workMenu.getWorkType().equals(WorkFlowMenuType.SUB_DIR.getCode())
                ||workMenu.getWorkType().equals(WorkFlowMenuType.SUB_SUB_DIR.getCode())){
            //验证对应的用户是否可以操作这个工程
            cheackProcId(workMenuDto.getProcId(), ShiroUtils.getUserId(), true, workMenu.getCreatePer(),"user",null);
        }else{
            //验证对应的用户是否可以操作这个工程
            cheackProcId(workMenuDto.getProcId(), ShiroUtils.getUserId(), true, workMenu.getCreatePer(),"admin",null);
        }


        //验证这个目录菜单是否可以被删除
        Map<String, Set<Long>> allChildrens = cheackMenu(workMenu);

        //删除对应的目录
        delMenu(workMenu, allChildrens);

    }

    @Override
    public WorkMenu addWorkFlowMenu(WorkMenuDto workMenuDto) {
        //验证参数
        if (null == workMenuDto) {
            throw new RRException(DataDevelopmentBizExceptionEnum.PARAM_ERROR.getMessage());
        }
        //验证需要新增的子目录
        if (null != workMenuDto.getParentId() && 0L != workMenuDto.getParentId()) {
            throw new RRException("该类型不是根目录类型!");
        }
        //名字必须传
        if (StringUtils.isEmpty(workMenuDto.getName())) {
            throw new RRException("目录名字必须传递!");
        }
        if (null == workMenuDto.getProcId()) {
            throw new RRException("工程id不能为空!");
        }
        //验证对应的用户是否可以操作这个工程
        cheackProcId(workMenuDto.getProcId(), ShiroUtils.getUserId(), false, null,null,null);

        //验证当前工程中的名字
        cheackProcIdAndName(workMenuDto.getProcId(), workMenuDto.getParentId(), workMenuDto.getName(), null);

        //新增目录
        return saveMenu(workMenuDto);
    }

    @Override
    public WorkMenu addRootWorkFlowMenu(WorkMenuDto workMenuDto) {
        //验证参数
        if (null == workMenuDto) {
            throw new RRException(DataDevelopmentBizExceptionEnum.PARAM_ERROR.getMessage());
        }
        //验证需要新增的子目录
        if (null != workMenuDto.getParentId() && 0L != workMenuDto.getParentId()) {
            throw new RRException("该类型不是根目录类型!");
        }
        //名字必须传
        if (StringUtils.isEmpty(workMenuDto.getName())) {
            throw new RRException("目录名字必须传递!");
        }
        if (null == workMenuDto.getProcId()) {
            throw new RRException("工程id不能为空!");
        }
        //验证对应的用户是否可以操作这个工程
        cheackProcId(workMenuDto.getProcId(), null == workMenuDto.getCreatePer() ? ShiroUtils.getUserId() : Long.valueOf(workMenuDto.getCreatePer()), false, null,null,null);

        //验证当前工程中的名字
        cheackProcIdAndName(workMenuDto.getProcId(), workMenuDto.getParentId(), workMenuDto.getName(), null);

        //新增根目录
        WorkMenu workMenu = new WorkMenu();
        BeanUtils.copyProperties(workMenuDto, workMenu);
        //如果是根目录的话，那么就默认是工作流程目录类型
        workMenu.setWorkType(WorkFlowMenuType.ROOT_DIR.getCode());
        workMenu.setCreatePer(null == workMenuDto.getCreatePer() ? ShiroUtils.getUserId().intValue() : workMenuDto.getCreatePer());
        workMenuMapper.insert(workMenu);
        workMenu.setRootId(workMenu.getId());
        workMenuMapper.updateById(workMenu);
        return workMenu;
    }

    @Override
    public WorkMenu addWorkFlowMGTAndSubMenu(WorkMenuDto workMenuDto) {
        //验证参数
        if (null == workMenuDto) {
            throw new RRException(DataDevelopmentBizExceptionEnum.PARAM_ERROR.getMessage());
        }
        //验证需要新增的子目录
        if (null == workMenuDto.getParentId() || 0L == workMenuDto.getParentId()) {
            throw new RRException("不能新增根目录!");
        }
        //名字必须传
        if (StringUtils.isEmpty(workMenuDto.getName())) {
            throw new RRException("目录名字必须传递!");
        }
        if (!Objects.equals(0L, workMenuDto.getParentId())) {
            WorkMenu workMenu = workMenuMapper.selectById(workMenuDto.getParentId());
            if (null == workMenu) {
                throw new RRException("父级工作流不存在!");
            }
            //验证是否可以添加目录
            if (!Objects.equals(workMenu.getWorkType(), WorkFlowMenuType.ROOT_DIR.getCode())
                    && !Objects.equals(workMenu.getWorkType(), WorkFlowMenuType.WORKFLOW_MGT.getCode())
                    && !Objects.equals(workMenu.getWorkType(), WorkFlowMenuType.SUB_DIR.getCode())
                    && !Objects.equals(workMenu.getWorkType(), WorkFlowMenuType.SUB_SUB_DIR.getCode())) {
                throw new RRException("父级菜单不能添加目录!");
            }
            workMenuDto.setRootId(workMenu.getRootId());
        }

        if (null == workMenuDto.getProcId()) {
            throw new RRException("工程id不能为空!");
        }
        //验证对应的用户是否可以操作这个工程
        cheackProcId(workMenuDto.getProcId(), null == workMenuDto.getCreatePer() ? ShiroUtils.getUserId() : Long.valueOf(workMenuDto.getCreatePer()), false, null,null,null);

        //验证当前工程中的名字
        cheackProcIdAndName(workMenuDto.getProcId(),workMenuDto.getParentId(), workMenuDto.getName(), null);

        WorkMenu workMenu = new WorkMenu();
        BeanUtils.copyProperties(workMenuDto, workMenu);
        workMenu.setCreatePer(null == workMenuDto.getCreatePer() ? ShiroUtils.getUserId().intValue() : workMenuDto.getCreatePer());
        workMenuMapper.insert(workMenu);
        return workMenu;
    }

    @Override
    public List<WorkMenuDto> listWorkFlowMenues(Integer procId) {
        //根据类型查询到对应的结构
        List<WorkMenu> workMenus = listByProcId(procId);
        if (CollectionUtil.isEmpty(workMenus)) {
            return new ArrayList<>();
        }

        List<JobLink> jobLinks = workMenuJobRelMapper.getJobLinks();
        if(jobLinks == null)
            jobLinks = new ArrayList<>();
        List<WorkMenuJobRel> jobRels = workMenuJobRelMapper.selectList(new QueryWrapper<WorkMenuJobRel>().eq("proc_id", procId));
        if(jobRels == null)
            jobRels = new ArrayList<>();
        List<JobNodeConf> jobNodeConfs = workMenuJobRelMapper.getJobNodeConfs(procId,null);
        if(jobNodeConfs == null)
            jobNodeConfs = new ArrayList<>();
        List<JobNodeInfo> jobNodeInfos = workMenuJobRelMapper.getJobNodeInfos(procId,null);
        if(jobNodeInfos == null)
            jobNodeInfos = new ArrayList<>();
        List<JobInfo> jobInfos = workMenuJobRelMapper.getJobInfos(procId,null);
        if(jobInfos == null)
            jobInfos = new ArrayList<>();

        //判断是否是可以权限转让，可以显示的---专门用于权限转让
        Long userId = ShiroUtils.getUserId();
        String transferCode = userId.toString();
        List<ProcUser> procUsers = procUserMapper.selectList(new QueryWrapper<ProcUser>().eq("proc_id", procId).eq("user_id", userId));
        if(DataPermissionUtil.isAdmin(userId.intValue()) || (CollectionUtil.isNotEmpty(procUsers) && Objects.equals(1,procUsers.get(0).getIsOwner()))){
            transferCode = "ALL";
        }
        //得到对应的数据
        List<WorkMenuDto> restList = null;
        try {
            restList = changeTreePack(PojoUtils.listConvert(WorkMenuDto.class, workMenus), false,true, jobLinks, jobRels, jobNodeConfs, jobNodeInfos, jobInfos,transferCode);
        } catch (IllegalAccessException e) {
            log.error("获取工作流菜单树异常，error:{}", e.getMessage());
            throw new RRException("获取工作流菜单树异常！");
        }
        if(CollectionUtil.isEmpty(restList)){
            return new ArrayList<>();
        }
        return restList;
    }



    @Override
    public List<WorkMenuDto> getWorkFlowMenusByUserId(Integer procId,Long userId) {

        //根据类型查询到对应的结构
        Map<String,Object> params = new HashMap<>();
        params.put("procId", procId);
        List<WorkMenu> workMenus = listByProcId(procId);
        Boolean flag = false;
        if (userId != null){
            if(userId == -1){
                userId = ShiroUtils.getUserId();
            }
            flag = true;
        }

        if (CollectionUtil.isEmpty(workMenus)) {
            return new ArrayList<>();
        }

        List<WorkMenuJobRel> jobRels = workMenuJobRelMapper.selectList(new QueryWrapper<WorkMenuJobRel>().eq("proc_id", procId));
        if(jobRels == null)
            jobRels = new ArrayList<>();
        List<JobNodeConf> jobNodeConfs = workMenuJobRelMapper.getJobNodeConfs(procId,null);
        if(jobNodeConfs == null)
            jobNodeConfs = new ArrayList<>();
        List<JobNodeInfo> jobNodeInfos = workMenuJobRelMapper.getJobNodeInfos(procId,null);
        if(jobNodeInfos == null)
            jobNodeInfos = new ArrayList<>();
        List<JobInfo> jobInfos = workMenuJobRelMapper.getJobInfos(procId,null);
        if(jobInfos == null)
            jobInfos = new ArrayList<>();

        //判断是否是可以权限转让，可以显示的---专门用于权限转让
        if (userId == null) userId = ShiroUtils.getUserId();
        String transferCode = userId.toString();
        List<ProcUser> procUsers = procUserMapper.selectList(new QueryWrapper<ProcUser>().eq("proc_id", procId).eq("user_id", userId));
        if(DataPermissionUtil.isAdmin(userId.intValue()) || (CollectionUtil.isNotEmpty(procUsers) && Objects.equals(1,procUsers.get(0).getIsOwner()))){
            transferCode = "ALL";
        }
        //得到对应的数据
        List<WorkMenuDto> restList = null;
        try {
            restList = changeTreePack(PojoUtils.listConvert(WorkMenuDto.class, workMenus), false,true, new ArrayList<>(), jobRels, jobNodeConfs, jobNodeInfos, jobInfos,transferCode);
        } catch (IllegalAccessException e) {
            log.error("获取工作流菜单树异常，error:{}", e.getMessage());
            throw new RRException("获取工作流菜单树异常！");
        }
        if(CollectionUtil.isEmpty(restList)){
            return new ArrayList<>();
        }

        if (flag) {//查询选择用户,含有任务的数据

            //存放该用户创建的子目录或者工作流
            List<WorkMenuDto> restListNew = new ArrayList<>();

            //存放已经添加到restListNew的子目录或者工作流的id
            Set<Long> menuIdList = new HashSet<>();

            //第一次迭代，获取restListNew数据，清洗不属于自己的数据
//            log.info("======================================restList.size=" + restList.size());
            Iterator<WorkMenuDto> iterator = restList.iterator();
            while (iterator.hasNext()){
                WorkMenuDto dto = iterator.next();
                //筛选该用户的数据
                changeOwnerData(userId,dto,restListNew,menuIdList);
                if (dto.getCreatePer() == Integer.parseInt(userId + "")) continue;
                if (!dto.getIsExist()) iterator.remove();
            }
            //第二次迭代，清洗不属于自己的数据
//            log.info("======================================restListNew.size=" + restListNew.size());
            restList.addAll(restListNew);
            Iterator<WorkMenuDto> iteratorNew = restList.iterator();
            while (iteratorNew.hasNext()){
                WorkMenuDto dto = iteratorNew.next();
                if (dto.getCreatePer() != Integer.parseInt(userId + "")) iteratorNew.remove();
//                if (!dto.getIsExist()) iteratorNew.remove();
            }
        }
//        log.info("======================================lastRestList.size=" + restList.size());
        return restList;
    }

    private void changeOwnerData(Long userId,WorkMenuDto workMenuDto,List<WorkMenuDto> restList,Set<Long> menuIdList) {

        List<WorkMenuDto> dtoChildrenList = (List<WorkMenuDto>)workMenuDto.getChildren();
        if (dtoChildrenList == null) return;

        if (dtoChildrenList.size() > 0) {
            //用于标记是否在该dtoChildrenList的循环中，有属于该用户的数据，有则标记后存parentId到menuIdList
            boolean flag = false;
            for (int i = 0; i < dtoChildrenList.size(); i++) {
                WorkMenuDto menuDto = dtoChildrenList.get(i);
//                if (menuDto.getWorkType() == null) continue;
                //判断属于该用户且WorkType=2的数据，是则将parentId保存到menuIdList；用于避免重复数据
                if (menuDto.getWorkType() != null && (menuDto.getWorkType() == 1 || menuDto.getWorkType() == 2)
                        && userId.equals(Long.parseLong(menuDto.getCreatePer() + ""))
                        && userId.equals(Long.parseLong(workMenuDto.getCreatePer() + ""))) menuIdList.add(menuDto.getId());
                //判断属于该用户且WorkType=3或5的数据，是则将id保存到menuIdList；用于避免重复数据;将整个menuDto存到restList
                if ((menuDto.getWorkType() == null || (menuDto.getWorkType() != 1 && menuDto.getWorkType() != 2)) && userId.equals(Long.parseLong(menuDto.getCreatePer() + ""))) {
                    if (menuDto.getIsExist()) workMenuDto.setIsExist(true);
                    if (!menuIdList.contains(menuDto.getParentId())) {
                        restList.add(menuDto);
                        flag = true;
                    }
                    menuIdList.add(menuDto.getId());
                    continue;
                }
                if (i == dtoChildrenList.size() - 1 && flag) menuIdList.add(menuDto.getParentId());
                changeOwnerData(userId, menuDto, restList, menuIdList);
                if (menuDto.getIsExist()) workMenuDto.setIsExist(true);
            }
        }

    }


    @Override
    public List<WorkSqlVO> taskSqls(Integer tenantId) {
        if(tenantId == null){
            tenantId = ShiroUtils.getTenantId();
        }
        Long userId = ShiroUtils.getUserId();
        //如果租户还不存在（平台超级管理员），则查询所有的数据
        List<WorkSqlVO> allTasks = new ArrayList<>();
        //1.离线任务
        if(tenantId == null){
            userId = null;
        }
        List<WorkSqlVO> tmpTasks =jobInfoMapper.tasks(userId);
        if(CollectionUtil.isNotEmpty(tmpTasks)){
            allTasks.addAll(tmpTasks);
        }
        //2.实时任务
        List<WorkSqlVO> tmpRealTasks =realTimeTaskMapper.tasks(userId);
        if(CollectionUtil.isNotEmpty(tmpRealTasks)){
            allTasks.addAll(tmpRealTasks);
        }
        return allTasks;
    }

    @Override
    public void pushWebData(Long jobMenuId, JobInfoDto jobInfoDto, List<JobNodeInfoDto> retJobNodeInfoList) {
        //webscoket推送状态
        WorkFlowStatusVO workFlowStatusVO = new WorkFlowStatusVO();
        workFlowStatusVO.setId(Long.valueOf(jobInfoDto.getId()));
        workFlowStatusVO.setStatus(jobInfoDto.getLastRunState());
        //查询到其下都有的节点，都是未执行
        if (CollectionUtil.isNotEmpty(retJobNodeInfoList)) {
            List<WorkFlowStatusVO> nodeList = new ArrayList<>();
            for (JobNodeInfoDto jobNodeInfo : retJobNodeInfoList) {
                WorkFlowStatusVO workFlowStatusVO1 = new WorkFlowStatusVO();
                workFlowStatusVO1.setId(Long.valueOf(jobNodeInfo.getId()));
                workFlowStatusVO1.setStatus(LastRunState.NOTRUN.getCode());
                nodeList.add(workFlowStatusVO1);
            }
            workFlowStatusVO.setNodeList(nodeList);
        }
        pushWebStatus(jobMenuId,workFlowStatusVO);
    }


    @Override
    public List<WorkMenuDto> listSeeWorkFlowMenues(Integer procId) {
        //如果是超级管理员和项目管理员可以看到项目下所有的工作流和工作流依赖
        List<WorkMenu> workMenus = listByProcId(procId);
        if (CollectionUtil.isEmpty(workMenus)) {
            return new ArrayList<>();
        }
        //工作流
        List<JobInfo> jobInfos = jobInfoMapper.selectList(new LambdaQueryWrapper<JobInfo>().eq(JobInfo::getProcId,procId));
        //工作流依赖
        List<WorkFlowDep> workFlowDeps = workFlowDepMapper.selectList(new LambdaQueryWrapper<WorkFlowDep>().eq(WorkFlowDep::getProcId, procId));
        //关联关系
        List<WorkMenuJobRel> jobRels = workMenuJobRelMapper.selectList(new QueryWrapper<WorkMenuJobRel>().eq("proc_id", procId));
        Long userId = ShiroUtils.getUserId();
        List<ProcUser> procUsers = procUserMapper.selectList(new QueryWrapper<ProcUser>().eq("proc_id", procId).eq("user_id", userId));
        Set<WorkMenuDto> needRestLists = new HashSet<>();
        if(DataPermissionUtil.isAdmin(userId.intValue()) || (CollectionUtil.isNotEmpty(procUsers) && Objects.equals(1,procUsers.get(0).getIsOwner()))){
            //递归到所有的数据
            if(CollectionUtil.isNotEmpty(jobInfos)){
                for (JobInfo jobInfo : jobInfos) {
                    if(CollectionUtil.isNotEmpty(jobRels)){
                        WorkMenuJobRel workMenuJobRel = jobRels.stream().filter(o -> (Objects.equals(jobInfo.getId(), o.getWorkFlowId().intValue()) && (Objects.equals(1, o.getWorkType())))).findAny().orElse(null);
                        if(null != workMenuJobRel){
                            doAllJobParents(workMenuJobRel.getWorkMenuId(),workMenus,needRestLists);
                        }
                    }
                }
            }
            if(CollectionUtil.isNotEmpty(workFlowDeps)){
                for (WorkFlowDep workFlowDep : workFlowDeps) {
                    doAllJobParents(workFlowDep.getWorkMenuId(),workMenus,needRestLists);
                }
            }
        }else{
            //只挑选出自己的工作流和依赖
            if(CollectionUtil.isNotEmpty(jobInfos)){
                List<JobInfo> owls = jobInfos.stream().filter(o -> Objects.equals(o.getCreatePer(), userId.intValue())).collect(Collectors.toList());
                if(CollectionUtil.isNotEmpty(owls)){
                    for (JobInfo jobInfo : owls) {
                        if(CollectionUtil.isNotEmpty(jobRels)){
                            WorkMenuJobRel workMenuJobRel = jobRels.stream().filter(o -> (Objects.equals(jobInfo.getId(), o.getWorkFlowId().intValue()) && (Objects.equals(1, o.getWorkType())))).findAny().orElse(null);
                            if(null != workMenuJobRel){
                                doAllJobParents(workMenuJobRel.getWorkMenuId(),workMenus,needRestLists);
                            }
                        }
                    }
                }
            }
            if(CollectionUtil.isNotEmpty(workFlowDeps)){
                List<WorkFlowDep> owls = workFlowDeps.stream().filter(o -> Objects.equals(o.getCreatePer(), userId.intValue())).collect(Collectors.toList());
                if(CollectionUtil.isNotEmpty(owls)){
                    for (WorkFlowDep workFlowDep : owls) {
                        doAllJobParents(workFlowDep.getWorkMenuId(),workMenus,needRestLists);
                    }
                }
            }
        }
        //封装树形结构
        List<WorkMenuDto> listTrees = Lists.newArrayList();
        List<WorkMenuDto> parentList = Lists.newArrayList();
        if (CollectionUtil.isNotEmpty(needRestLists)) {
            for (WorkMenuDto workMenuDto : needRestLists) {
                WorkMenuDto tempWorkMenu = new WorkMenuDto();
                BeanUtils.copyProperties(workMenuDto,tempWorkMenu);
                listTrees.add(tempWorkMenu);
                if (Objects.equals(0L,workMenuDto.getParentId())) {
                    parentList.add(tempWorkMenu);
                }
            }
            Map<Long, List<WorkMenuDto>> sysListMap = listTrees.stream().filter(s -> !Objects.equals(0L,s.getParentId()))
                    .collect(Collectors.groupingBy(WorkMenuDto::getParentId));
            listTrees.forEach(s -> s.setChildren(sysListMap.get(s.getId())));
        }
        return parentList;
    }

    private void doAllJobParents(Long workMenuId, List<WorkMenu> workMenus, Set<WorkMenuDto> needRestLists) {
        if(CollectionUtil.isEmpty(workMenus)){
            return;
        }
        WorkMenu workMenu = workMenus.stream().filter(o -> Objects.equals(o.getId(), workMenuId)).findAny().orElse(null);
        if(null != workMenu){
            WorkMenuDto workMenuDto = new WorkMenuDto();
            BeanUtils.copyProperties(workMenu,workMenuDto);
            needRestLists.add(workMenuDto);
            needRestLists.add(workMenuDto);
            doAllJobParents(workMenu.getParentId(),workMenus,needRestLists);
        }
    }

    private  List<WorkMenu> listByProcId(Integer procId){
        Map<String,Object> params = new HashMap<>();
        params.put("procId", procId);
        return workMenuMapper.listBy(params);
    }

    @Override
    public List<WorkMenuDto> listWorkFlowMenuesNotIncludeNodes(Integer procId) {
        //根据类型查询到对应的结构
        List<WorkMenu> workMenus = workMenuMapper.selectList(new QueryWrapper<WorkMenu>()
                .eq("proc_id", procId));
        if (CollectionUtil.isEmpty(workMenus)) {
            return null;
        }
        //得到对应的用户名称
        for (WorkMenu workMenu : workMenus) {
            workMenu.setCreatePerName(userNameByUserId(workMenu.getCreatePer()));
        }
        //得到对应的数据
        List<WorkMenuDto> restList = null;
        try {
            restList = changeTreePack(PojoUtils.listConvert(WorkMenuDto.class, workMenus), false,false);
        } catch (IllegalAccessException e) {
            log.error("获取工作流菜单树异常，error:{}", e.getMessage());
            throw new RRException("获取工作流菜单树异常！");
        }
        return restList;
    }

    @Override
    public void updateWorkFlowMenuName(WorkMenuDto workMenuDto) {
        if (null == workMenuDto) {
            throw new RRException(DataDevelopmentBizExceptionEnum.PARAM_ERROR.getMessage());
        }
        if (null == workMenuDto.getProcId()) {
            throw new RRException("工程id不能为空!");
        }
        WorkMenu workMenu = workMenuMapper.selectById(workMenuDto.getId());
        if (null == workMenu) {
            throw new RRException("目录不存在!");
        }

        //验证对应的用户是否可以操作这个工程
        if (workMenu.getWorkType().equals(WorkFlowMenuType.ROOT_DIR.getCode())
                ||workMenu.getWorkType().equals(WorkFlowMenuType.SUB_DIR.getCode())
                ||workMenu.getWorkType().equals(WorkFlowMenuType.SUB_SUB_DIR.getCode())
        ){
            //验证对应的用户是否可以操作这个工程
            cheackProcId(workMenuDto.getProcId(), ShiroUtils.getUserId(), true, workMenu.getCreatePer(),"user",null);
        }else{
            //验证对应的用户是否可以操作这个工程
            cheackProcId(workMenuDto.getProcId(), ShiroUtils.getUserId(), true, workMenu.getCreatePer(),"admin",null);
        }

        if (null == workMenuDto.getId()) {
            throw new RRException("目录id必须传!");
        }

        //名字必须传
        if (StringUtils.isEmpty(workMenuDto.getName())) {
            throw new RRException("目录名字必须传递!");
        }

        //验证当前工程中的名字
        cheackProcIdAndName(workMenuDto.getProcId(), workMenuDto.getParentId(), workMenuDto.getName(), workMenuDto.getId());

        workMenu.setName(workMenuDto.getName());
        workMenuMapper.updateById(workMenu);

        //默认工作流文件夹
        if(workMenu.getWorkType() == WorkFlowMenuType.WORKFLOW_FIELD.getCode()){
            WorkMenuJobRel workMenuJobRel = workMenuJobRelMapper.selectOne(new QueryWrapper<WorkMenuJobRel>().eq("work_menu_id", workMenu.getId()).eq("work_type", 1));
            if(null != workMenuJobRel){
                JobInfo jobInfo = jobInfoMapper.selectById(workMenuJobRel.getWorkFlowId());
                if(null != jobInfo){
                    jobInfo.setName(workMenuDto.getName());
                    jobInfoMapper.updateById(jobInfo);
                }
            }
        }
    }

    @Override
    public void delWorkFlow(WorkMenuDto workMenuDto) {
        //查看当前节点是否正在运行中
        if (null == workMenuDto) {
            throw new RRException(DataDevelopmentBizExceptionEnum.PARAM_ERROR.getMessage());
        }
        if (null == workMenuDto.getProcId()) {
            throw new RRException("工程id不能为空!");
        }
        if (null == workMenuDto.getId()) {
            throw new RRException("任务节点id必须传!");
        }
        //查询对应的节点是否存在
        JobNodeInfo jobNodeInfo = jobNodeInfoMapper.selectById(workMenuDto.getId());
        if (null == jobNodeInfo) {
            throw new RRException("任务节点不存在!");
        }
        if (null != jobNodeInfo.getLastRunState() && LastRunState.RUNNING.getCode() == jobNodeInfo.getLastRunState()) {
            throw new RRException("任务节点正在运行，请稍后重试!");
        }

        //验证对应的用户是否可以操作这个工程
        cheackProcId(workMenuDto.getProcId(), ShiroUtils.getUserId(), true, jobNodeInfo.getCreatePer(),"admin",null);


//        //当前节点是否在画布中存在，如果存在也不能删除---这个已经不能了，得修改。已经没有nodeKey了
//        cheackPointJobNode(jobNodeInfo);
//
//        //删除对应的节点数据---这个已经不能了，得修改。已经没有nodeKey了
//        delWorkFlowLinked(jobNodeInfo);

        //当前节点在画布中是否存在，如果存在，则不能删除
        cheackPointJobNodeNew(jobNodeInfo);

        //删除对应的节点数据
        delWorkFlowLinkedNew(jobNodeInfo);

        //删除模型任务表中的数据;
        bdpModelTaskMapper.delete(new QueryWrapper<BdpModelTask>().eq("task_id",workMenuDto.getId()));

        //删除配置信息
        jobNodeConfMapper.delete(new QueryWrapper<JobNodeConf>().eq("job_node_id", jobNodeInfo.getId()).eq("job_type",0));

    }

    private void delWorkFlowLinkedNew(JobNodeInfo jobNodeInfo) {
        if (null == jobNodeInfo) {
            return;
        }
        //删除对应的节点数据
        jobNodeInfoMapper.deleteById(jobNodeInfo.getId());
        workMenuJobRelMapper.delete(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 0)
                .eq("work_flow_id", jobNodeInfo.getId()));
    }

    private void cheackPointJobNodeNew(JobNodeInfo jobNodeInfo) {
        //验证画布中是否有配置这个节点，如果有的话，则不能删除
        if (null == jobNodeInfo) {
            return;
        }
        if (null != jobNodeInfo.getPosX() && null != jobNodeInfo.getPosY()) {
            throw new RRException("任务节点存在于画布中，请先移除任务节点!");
        }
        List<JobLink> jobLinks = jobLinkMapper.selectList(new QueryWrapper<JobLink>().eq("linke_type", 0).eq("dst_node_id", jobNodeInfo.getId()));
        List<JobLink> jobLinks1 = jobLinkMapper.selectList(new QueryWrapper<JobLink>().eq("linke_type", 0).eq("src_node_id", jobNodeInfo.getId()));
        if (CollectionUtil.isNotEmpty(jobLinks) || CollectionUtil.isNotEmpty(jobLinks1)) {
            throw new RRException("任务节点存在于画布中有连线，请先移除任务节点!");
        }
    }


    @Override
    public void updateWorkFlowName(WorkMenuDto workMenuDto) {
        if (null == workMenuDto) {
            throw new RRException(DataDevelopmentBizExceptionEnum.PARAM_ERROR.getMessage());
        }
        if (null == workMenuDto.getProcId()) {
            throw new RRException("工程id不能为空!");
        }
        JobNodeInfo jobNodeInfo = jobNodeInfoMapper.selectById(workMenuDto.getId());
        if (null == jobNodeInfo) {
            throw new RRException("任务节点不存在!");
        }
        //名字必须传
        if (StringUtils.isEmpty(workMenuDto.getName())) {
            throw new RRException("目录名字必须传递!");
        }

        //验证对应的用户是否可以操作这个工程
        cheackProcId(workMenuDto.getProcId(), ShiroUtils.getUserId(), true, jobNodeInfo.getCreatePer(),"admin",null);
        if (null == workMenuDto.getId()) {
            throw new RRException("节点id必须传!");
        }


        //验证当前工程中的名字
//        cheackProcIdAndNameJob(workMenuDto.getProcId(),workMenuDto.getName(),workMenuDto.getId());

        jobNodeInfo.setName(workMenuDto.getName());
        jobNodeInfoMapper.updateById(jobNodeInfo);
    }

    @Override
    public void updateShellJob(WorkMenuDto workMenuDto) {
        if (null == workMenuDto) {
            throw new RRException(DataDevelopmentBizExceptionEnum.PARAM_ERROR.getMessage());
        }
        if (null == workMenuDto.getProcId()) {
            throw new RRException("工程id不能为空!");
        }
        JobNodeInfo jobNodeInfo = jobNodeInfoMapper.selectById(workMenuDto.getId());
        if (null == jobNodeInfo) {
            throw new RRException("任务节点不存在!");
        }
        //名字必须传
        if (StringUtils.isEmpty(workMenuDto.getName())) {
            throw new RRException("目录名字必须传递!");
        }
        //验证对应的用户是否可以操作这个工程
        cheackProcId(workMenuDto.getProcId(), ShiroUtils.getUserId(), true, jobNodeInfo.getCreatePer(),"admin",null);
        if (null == workMenuDto.getId()) {
            throw new RRException("节点id必须传!");
        }
        jobNodeInfo.setName(workMenuDto.getName());
        jobNodeInfo.setDesc(workMenuDto.getDescName());
        jobNodeInfoMapper.updateById(jobNodeInfo);

        //编辑对应的数据
        BussessConfigEntity bussessConfigEntity = bussessConfigMapper.selectOne(new QueryWrapper<BussessConfigEntity>()
                .eq("bussess_type", 1).eq("bussess_id", jobNodeInfo.getId()));
        if(null != bussessConfigEntity){
            bussessConfigMapper.deleteById(bussessConfigEntity);
        }
        BussessConfigEntity newBussessConfigEntity = new BussessConfigEntity();
        newBussessConfigEntity.setBussessType(1);
        newBussessConfigEntity.setBussessId(jobNodeInfo.getId());
        newBussessConfigEntity.setHostUrl(workMenuDto.getHostUrl());
        newBussessConfigEntity.setPort(workMenuDto.getPort());
        newBussessConfigEntity.setUsername(workMenuDto.getUsername());
        newBussessConfigEntity.setPassword(workMenuDto.getPassword());
        newBussessConfigEntity.setCreateTime(new Date());
        bussessConfigMapper.insert(newBussessConfigEntity);

    }

    @Override
    public Map<String, JSONObject> updateInitWorkflowNodeParams() {
        //查询到所有的数据 sysParam  -> nodeParam
        List<JobNodeConf> jobNodeConfs = jobNodeConfMapper.selectList(new QueryWrapper<JobNodeConf>()
                .eq("job_type", 1)
                .eq("`key`",
                        "sysParam"));
        List<JobNodeConf> newSysParams = new ArrayList<>();
        if(CollectionUtil.isNotEmpty(jobNodeConfs)){
            Map<String, JSONObject> restListMaps = new HashMap<>();
            for (JobNodeConf jobNodeConf : jobNodeConfs) {
                if(StringUtils.isNotEmpty(jobNodeConf.getValue())){
                    JSONObject lll = new JSONObject();
                    String value = jobNodeConf.getValue();
                    lll.put("olderValues",value);
                    JSONArray jsonArray = JSONArray.parseArray(value);
                    if(CollectionUtil.isNotEmpty(jsonArray)){
                        List<JSONObject> lllJs = new ArrayList<>();
                        for (Object o : jsonArray) {
                            if(null != o){
                                JSONObject jsonObject = JSONObject.parseObject(o.toString());
                                if(null != jsonObject){
                                    Map<String, Object> innerMap = jsonObject.getInnerMap();
                                    if(null != innerMap){
                                        Set<String> strings = innerMap.keySet();
                                        if(CollectionUtil.isNotEmpty(strings)){
                                            JSONObject obj = new JSONObject();
                                            Object o1 = strings.toArray()[0];
                                            obj.put("key",o1);
                                            obj.put("value",innerMap.get(o1));
                                            lllJs.add(obj);
                                        }
                                    }
                                }
                            }
                        }
                        if(CollectionUtil.isNotEmpty(lllJs)){
                            String s = JSONArray.toJSONString(lllJs);
                            lll.put("newVaules",s);
                            JobNodeConf jobNodeConf1 = new JobNodeConf();
                            BeanUtils.copyProperties(jobNodeConf,jobNodeConf1);
                            jobNodeConf1.setId(null);
                            jobNodeConf1.setKey("nodeParam");
                            jobNodeConf1.setValue(s);
                            newSysParams.add(jobNodeConf1);
                            restListMaps.put(jobNodeConf.getJobNodeId().toString(),lll);
                        }
                    }
                }
            }
            //修改所有的数据格式
            if(CollectionUtil.isNotEmpty(newSysParams)){
                jobNodeConfMapper.upsertKVByJobNodeId(newSysParams);
            }
            return restListMaps;
        }
        return null;
    }


    @Override
    public WorkMenu addDefaultWorkFlowMenu(WorkMenuDto workMenuDto) {
        //验证参数
        if (null == workMenuDto) {
            throw new RRException(DataDevelopmentBizExceptionEnum.PARAM_ERROR.getMessage());
        }
        if (null == workMenuDto.getParentId() || 0L == workMenuDto.getParentId()) {
            throw new RRException("父级目录不能为空!");
        }
        WorkMenu workMenu = workMenuMapper.selectById(workMenuDto.getParentId());
        if (null == workMenu) {
            throw new RRException("父级目录不存在!");
        }
        if (WorkFlowMenuType.SUB_DIR.getCode() != workMenu.getWorkType() && WorkFlowMenuType.SUB_SUB_DIR.getCode() != workMenu.getWorkType()) {
            throw new RRException("当前父级目录下不能新建默认作业流类型的文件夹!");
        }

        //验证是否可以添加默认作业流
        if (null == workMenuDto.getProcId()) {
            throw new RRException("工程id不能为空!");
        }
        //验证对应的用户是否可以操作这个工程
        cheackProcId(workMenuDto.getProcId(), null == workMenuDto.getCreatePer() ? ShiroUtils.getUserId() : Long.valueOf(workMenuDto.getCreatePer()), false, null,null,null);

        //名字必须传
        if (StringUtils.isEmpty(workMenuDto.getName())) {
            throw new RRException("目录名字必须传递!");
        }

        //验证当前工程中的名字
        cheackProcIdAndName(workMenuDto.getProcId(), workMenuDto.getParentId(), workMenuDto.getName(), null);

        //添加作业流文件夹，并且添加
        WorkMenu workMenu1 = new WorkMenu();
        workMenu1.setProcId(workMenuDto.getProcId());
        workMenu1.setParentId(workMenuDto.getParentId());
        workMenu1.setWorkType(WorkFlowMenuType.WORKFLOW_FIELD.getCode());
        workMenu1.setName(workMenuDto.getName());
        workMenu1.setCreateTime(LocalDateTime.now());
        workMenu1.setRootId(workMenu.getRootId());
        workMenu1.setCreatePer(null == workMenuDto.getCreatePer() ? ShiroUtils.getUserId().intValue() : workMenuDto.getCreatePer());
        if(null != workMenuDto.getId() && !Objects.equals(0L,workMenuDto.getId())){
            workMenu1.setId(workMenuDto.getId());
            workMenuMapper.insertById(workMenu1);
        }else{
            workMenuMapper.insert(workMenu1);
        }


        //新增job名字
        JobInfo jobInfo = new JobInfo();
        jobInfo.setName(workMenuDto.getName());
        jobInfo.setEnable(JobStatus.ACTIVE.getCode());
        jobInfo.setCreateTime(DateTime.now());
        jobInfo.setCreatePer(null == workMenuDto.getCreatePer() ? ShiroUtils.getUserId().intValue() : workMenuDto.getCreatePer());
        jobInfo.setProcId(workMenuDto.getProcId());
        jobInfo.setParam(null);
        jobInfo.setDesc(workMenuDto.getDescName());
        if(null != workMenuDto.getOlderJobId()){
            jobInfo.setId(workMenuDto.getOlderJobId());
            jobInfoMapper.insertJobInfo(jobInfo);
        }else{
            jobInfoMapper.insert(jobInfo);
        }

        //任务和菜单的关联关系数据
        WorkMenuJobRel workMenuJobRel = new WorkMenuJobRel();
        workMenuJobRel.setProcId(workMenuDto.getProcId());
        workMenuJobRel.setWorkType(1);
        workMenuJobRel.setWorkMenuId(workMenu1.getId());
        workMenuJobRel.setWorkFlowId(Long.valueOf(jobInfo.getId()));
        workMenuJobRel.setCreateTime(LocalDateTime.now());
        workMenuJobRelMapper.insert(workMenuJobRel);

        // 虚拟化一个开始节点和一个结束节点，并且保存到hdfs上
        saveStartNodeAndEndNode(workMenuDto.getProcId().longValue(), jobInfo.getId(), WorkFlowType.WORKFLOW.getCode());

        return workMenu1;


    }

    @Override
    public void saveStartNodeAndEndNode(Long procId, Integer jobId, Integer type) {
        JobNodeInfo jobNodeInfo = new JobNodeInfo();
        jobNodeInfo.setJobId(jobId);
        jobNodeInfo.setTypeId(0);
        String nodeName = JobType.ObjOf(0).getEnName();
        String dst = "workflow-app/" + type + "/" + procId + "/" + jobId.toString() + "/" + nodeName + "_0.sh";
        jobNodeInfoService.uploadShell(jobNodeInfo, dst, oozieConfig);

        JobNodeInfo jobNodeInfo1 = new JobNodeInfo();
        jobNodeInfo1.setJobId(jobId);
        jobNodeInfo1.setTypeId(5);
        String nodeName1 = JobType.ObjOf(5).getEnName();
        String dst1 = "workflow-app/" + type + "/" + +procId + "/" + jobId.toString() + "/" + nodeName1 + "_0.sh";
        jobNodeInfoService.uploadShell(jobNodeInfo1, dst1, oozieConfig);

    }

    @Override
    public void saveStartJobNodeAndEndJobNode(Long procId, Integer menuDepId, Integer type) {
        String nodeName = JobType.ObjOf(0).getEnName();
        String dst = "workflow-app/" + type + "/" + procId + "/" + menuDepId + "/" + nodeName + "_0.sh";
        jobNodeInfoService.uploadJobShell(0, dst, oozieConfig,menuDepId);

        String nodeName1 = JobType.ObjOf(5).getEnName();
        String dst1 = "workflow-app/" + type + "/" + +procId + "/" + menuDepId + "/" + nodeName1 + "_0.sh";
        jobNodeInfoService.uploadJobShell(5, dst1, oozieConfig,menuDepId);

    }

    @Override
    public Integer addWorkJob(WorkMenuDto workMenuDto) {
        //验证参数
        if (null == workMenuDto) {
            throw new RRException(DataDevelopmentBizExceptionEnum.PARAM_ERROR.getMessage());
        }
        WorkMenu workMenu = workMenuMapper.selectById(workMenuDto.getParentId());
        if (null == workMenu) {
            throw new RRException("默认作业流不存在!");
        }
        if (WorkFlowMenuType.WORKFLOW_FIELD.getCode() != workMenu.getWorkType()) {
            throw new RRException("上级目录不是默认作业流类型!");
        }
        //验证是否可以添加默认作业流
        if (null == workMenuDto.getProcId()) {
            throw new RRException("工程id不能为空!");
        }
        //暂时不添加任务节点
//        if (!workMenu.getCreatePer().equals(ShiroUtils.getUserId().intValue())){
//            throw new RRException("只能在本人创建的作业流下面添加任务节点!");
//        }
        //验证对应的用户是否可以操作这个工程
        cheackProcId(workMenuDto.getProcId(), ShiroUtils.getUserId(), false, null,null,null);

        //名字必须传
        if (StringUtils.isEmpty(workMenuDto.getName())) {
            throw new RRException("目录名字必须传递!");
        }
        if (null == workMenuDto.getTypeId()) {
            throw new RRException("任务类型必须传递!");
        }

        //验证节点的名字是否重复
        WorkMenuJobRel workMenuJobRel = workMenuJobRelMapper.selectOne(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 1).eq("work_menu_id", workMenuDto.getParentId()));
        if (null == workMenuJobRel) {
            throw new RRException("任务和默认作业流菜单关联不存在!");
        }

        BussessConfigEntity bussessConfigEntity = null;
        if(Objects.equals(workMenuDto.getTypeId(),JobType.SHELL.getCode())){
            if(StringUtils.isEmpty(workMenuDto.getHostUrl())){
                throw new RRException("服务器地址未配置!");
            }
            if(null == workMenuDto.getPort()){
                throw new RRException("服务器端口未配置!");
            }
            if(StringUtils.isEmpty(workMenuDto.getUsername())){
                throw new RRException("服务器用户名未配置!");
            }
            if(StringUtils.isEmpty(workMenuDto.getPassword())){
                throw new RRException("服务器密码未配置!");
            }
            bussessConfigEntity = new BussessConfigEntity();
            bussessConfigEntity.setBussessType(1);
            bussessConfigEntity.setHostUrl(workMenuDto.getHostUrl());
            bussessConfigEntity.setPort(workMenuDto.getPort());
            bussessConfigEntity.setUsername(workMenuDto.getUsername());
            bussessConfigEntity.setPassword(workMenuDto.getPassword());
            bussessConfigEntity.setCreateTime(new Date());

        }

        JobNodeInfo jobNodeInfo = new JobNodeInfo();
        jobNodeInfo.setName(workMenuDto.getName());
        jobNodeInfo.setTypeId(workMenuDto.getTypeId());
        jobNodeInfo.setEnable(JobStatus.ACTIVE.getCode());
        jobNodeInfo.setCreateTime(DateTime.now());
        jobNodeInfo.setCreatePer(ShiroUtils.getUserId().intValue());
        jobNodeInfo.setJobId(workMenuJobRel.getWorkFlowId().intValue());
        jobNodeInfo.setNodeKey(null);
        jobNodeInfo.setPNodeKey(null);
        jobNodeInfo.setDesc(workMenuDto.getDescName());
        jobNodeInfoMapper.insert(jobNodeInfo);

        //添加shell脚本对应的配置信息
        if(null != bussessConfigEntity){
            bussessConfigEntity.setBussessId(jobNodeInfo.getId());
            bussessConfigMapper.insert(bussessConfigEntity);
        }

        //添加节点关联表数据
        WorkMenuJobRel workMenuJobRel1 = new WorkMenuJobRel();
        BeanUtils.copyProperties(workMenuJobRel, workMenuJobRel1);
        workMenuJobRel1.setId(null);
        workMenuJobRel1.setWorkFlowId(Long.valueOf(jobNodeInfo.getId()));
        workMenuJobRel1.setWorkType(0);
        workMenuJobRelMapper.insert(workMenuJobRel1);
        return jobNodeInfo.getId();
    }

    @Override
    public List<JobNodeInfoDto> listJobNodeList(WorkMenuDto workMenuDto) {
        //查询出来位置不为null的，则一定是有拖入画布的
        if (null == workMenuDto || workMenuDto.getId() == null) {
            throw new RRException(DataDevelopmentBizExceptionEnum.PARAM_ERROR.getMessage());
        }
        WorkMenu workMenu = workMenuMapper.selectById(workMenuDto.getId());
        if (null == workMenu) {
            throw new RRException("默认作业流不存在!");
        }
        if (WorkFlowMenuType.WORKFLOW_FIELD.getCode() != workMenu.getWorkType()) {
            throw new RRException("上级目录不是默认作业流类型!");
        }
        List<WorkMenuJobRel> workMenuJobRels = workMenuJobRelMapper.selectList(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 0).eq("work_menu_id", workMenuDto.getId()));
        Set<Integer> jobNodeIds = new HashSet<>();
        for (WorkMenuJobRel workMenuJobRel : workMenuJobRels) {
            jobNodeIds.add(workMenuJobRel.getWorkFlowId().intValue());
        }
        if (CollectionUtil.isEmpty(jobNodeIds)) {
            return null;
        }
        List<JobNodeInfoDto> listJobNodeInfoDtos = jobNodeInfoMapper.selectListByIds(jobNodeIds);
        if (CollectionUtil.isNotEmpty(listJobNodeInfoDtos)) {
            return listJobNodeInfoDtos;
        }
        return null;
    }

    @Override
    public List<JobLinkDto> listJobLinkList(WorkMenuDto workMenuDto, Integer linkeType) {
        //查询出来位置不为null的，则一定是有拖入画布的
        if (null == workMenuDto || workMenuDto.getId() == null) {
            throw new RRException(DataDevelopmentBizExceptionEnum.PARAM_ERROR.getMessage());
        }
        WorkMenu workMenu = workMenuMapper.selectById(workMenuDto.getId());
        if (null == workMenu) {
            throw new RRException("默认作业流不存在!");
        }
        if (0 == linkeType) {
            if (WorkFlowMenuType.WORKFLOW_FIELD.getCode() != workMenu.getWorkType()) {
                throw new RRException("上级目录不是默认作业流类型!");
            }
        }
        if (1 == linkeType) {
            if (WorkFlowMenuType.WORKFLOW_MGT_FIELD.getCode() != workMenu.getWorkType()) {
                throw new RRException("上级目录不是作业流血缘文件夹类型!");
            }
        }
        List<JobLinkDto> listJobLinkDtos = jobLinkMapper.selectListByMenuId(workMenu.getId(), linkeType);
        if (CollectionUtil.isNotEmpty(listJobLinkDtos)) {
            return listJobLinkDtos;
        }
        return null;
    }

    @Override
    public void updateWorkJob(JobLinkAndJobNodeDto jobLinkAndJobNodeDto) {

        if (null == jobLinkAndJobNodeDto.getProcId()) {
            throw new RRException("工程id不能为空!");
        }
        WorkMenu workMenu = workMenuMapper.selectById(jobLinkAndJobNodeDto.getWorkFlowMenuId());
        if (null == workMenu) {
            throw new RRException("默认作业流不存在!");
        }
        if (WorkFlowMenuType.WORKFLOW_FIELD.getCode() != workMenu.getWorkType()) {
            throw new RRException("上级目录不是默认作业流类型!");
        }
        //验证对应的用户是否可以操作这个工程
        cheackProcId(jobLinkAndJobNodeDto.getProcId(), ShiroUtils.getUserId(), true, workMenu.getCreatePer(),"admin",jobLinkAndJobNodeDto.getIsImport());

        //查询这个菜单对应的job任务
        List<WorkMenuJobRel> workMenuJobRels = workMenuJobRelMapper.selectList(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 1).eq("work_menu_id", workMenu.getId()));
        if (CollectionUtil.isEmpty(workMenuJobRels)) {
            throw new RRException("默认作业流对应的工作流不存在!");
        }
        //验证对应的任务的运行状态
        JobInfo jobInfo = jobInfoMapper.selectById(workMenuJobRels.get(0).getWorkFlowId());
        if (null == jobInfo) {
            throw new RRException("工作流不存在!");
        }
        if (null != jobInfo.getLastRunState() && jobInfo.getLastRunState() == LastRunState.RUNNING.getCode()) {
            throw new RRException("作业流正在运行中，请稍后重试!");
        }

        //引用了这个工作流（配置的自动功能变化为暂停调度,自动运行的调度给停止掉）的工作流依赖是否在运行中（是否是禁用还是启用还是在运行中）
        cheackWorkFlowJobLay(workMenuJobRels.get(0).getWorkFlowId(), jobLinkAndJobNodeDto);

        //验证表达式数字
        cheackExpressionNumber(jobLinkAndJobNodeDto.getJobNodeConfig());

        //验证连线和路由配置的条件的冲突情况
        cheackJobNodeLayConfig(jobLinkAndJobNodeDto.getNodes(), jobLinkAndJobNodeDto.getEdges(), workMenuJobRels.get(0).getWorkFlowId());

        //验证节点和连线信息
        Set<Integer> nodeIds = new HashSet<>();
        Set<JobNodeInfoDto> nodeDtos = new HashSet<>();
        if(CollectionUtil.isNotEmpty(jobLinkAndJobNodeDto.getNodes())){
            for (JobNodeInfoDto jobNodeInfoDto : jobLinkAndJobNodeDto.getNodes()) {
                nodeIds.add(jobNodeInfoDto.getId());
                JobNodeInfoDto newJobNodeDto = new JobNodeInfoDto();
                newJobNodeDto.setId(jobNodeInfoDto.getId());
                //是否配置空跑
                newJobNodeDto.setIsNullRun(jobNodeInfoDto.getIsNullRun());
                nodeDtos.add(newJobNodeDto);
            }
        }
        if(CollectionUtil.isEmpty(nodeIds)){

            if(null != jobLinkAndJobNodeDto.getJobNodeConfig() && null != jobLinkAndJobNodeDto.getJobNodeConfig().getSchedule_type() && jobLinkAndJobNodeDto.getJobNodeConfig().getSchedule_type() == 1){
                throw new RRException("未设置节点，不能设置成正常调度！");
            }

            //清空数据
            jobNodeInfoMapper.clearPosOthers(workMenuJobRels.get(0).getWorkFlowId());
            //更新连线和依赖
            changeJobNodeLayConfig(jobLinkAndJobNodeDto.getNodes(), jobLinkAndJobNodeDto.getEdges(), workMenuJobRels.get(0).getWorkFlowId());
            //保存基本属性配置
            saveScheduleData(jobLinkAndJobNodeDto.getJobNodeConfig(), jobLinkAndJobNodeDto.getWorkFlowMenuId(),true);

            //更新连线
            jobLinkMapper.delete(new QueryWrapper<JobLink>().eq("linke_type", 0).eq("menu_id", jobLinkAndJobNodeDto.getWorkFlowMenuId()));

            //停止任务并且删除文件目录
            //kill掉oozie中的job
            JobUtil jobUtil = null;
            HdfsUtil hdfsUtil = null;
            try {
                OozieClient wc = new OozieClient(oozieConfig.getUrl());
                hdfsUtil = new HdfsUtil(bdpJobConfig);
                jobUtil = new JobUtil(hdfsUtil, wc);
                if (StringUtils.isNotEmpty(jobInfo.getLastCorOozieJobId())) {
                    jobUtil.stopJob(jobInfo.getLastCorOozieJobId());
                    cheackJobStatusAndOozieId(wc,jobInfo.getLastCorOozieJobId());
                }
                if (StringUtils.isNotEmpty(jobInfo.getLastOozieJobId())) {
                    jobUtil.stopJob(jobInfo.getLastOozieJobId());
                    cheackJobStatusAndOozieId(wc,jobInfo.getLastOozieJobId());
                }
            } catch (Exception e) {
                log.error("停止任务失败。error:" + e.getMessage());
                throw new RRException("停止任务失败,"+e.getMessage());
            }finally {
                if(null != hdfsUtil){
                    hdfsUtil.close();
                }
            }
            //1.改变工作流的状态
            jobInfoService.updateStatus(jobInfo.getId());
            //2.暂停调度的工作流
            ExecutorService executorService = Executors.newSingleThreadExecutor();
            executorService.execute(()->{
                //暂停调度的工作流
                jobInfoService.stopAutoJob(jobInfo.getId());
            });

            deleteHdfs(workMenu.getProcId(),jobInfo.getId());

            return;
        }



        //默认添加了开始节点
        cheackLinked(jobLinkAndJobNodeDto.getEdges(), nodeIds);

        //验证节点是否全部都配置过配置,如果没有，则不知道如何进行保存
        cheackJobNodeIsConfig(jobLinkAndJobNodeDto.getNodes());

        //验证节点是否是属于这个工作流
        cheackJobNodeIsWorkFlowMenu(jobLinkAndJobNodeDto.getNodes(), workMenu.getId());


        //重新获取去掉开始节点
        List<JobLinkDto> insertDataList = new ArrayList<>();
        List<JobLinkDto> newInsertDataList = new ArrayList<>();
        if(CollectionUtil.isNotEmpty(jobLinkAndJobNodeDto.getEdges())){
            jobLinkAndJobNodeDto.getEdges().forEach(jobLinkDto -> {
                if (0 != jobLinkDto.getSrcNodeId()) {
                    JobLinkDto jobLinkDto1 = new JobLinkDto();
                    BeanUtils.copyProperties(jobLinkDto, jobLinkDto1);
                    jobLinkDto1.setCreatePer(ShiroUtils.getUserId().intValue());
                    jobLinkDto1.setCreateTime(DateTime.now());
                    jobLinkDto1.setLinkeType(0);
                    jobLinkDto1.setMenuId(jobLinkAndJobNodeDto.getWorkFlowMenuId());
                    insertDataList.add(jobLinkDto1);
                    newInsertDataList.add(jobLinkDto1);
                }
            });
        }
        //先查询出来对应的历史数据
        List<JobNodeConf> jobNodeConfs = listWorkFlowConfig(workMenu.getId());
        //保存基本属性配置
        saveScheduleData(jobLinkAndJobNodeDto.getJobNodeConfig(), jobLinkAndJobNodeDto.getWorkFlowMenuId(), true);

//        //是否刷新节点的shell文件
//        if(null != jobLinkAndJobNodeDto.getJobNodeConfig() && CollectionUtil.isNotEmpty(jobLinkAndJobNodeDto.getJobNodeConfig().getSysParam())){
//            try {
//                refreshNodeShell(jobLinkAndJobNodeDto.getJobNodeConfig().getSysParam(),nodeIds,jobLinkAndJobNodeDto.getIsImport());
//            } catch (Exception e) {
//                if(CollectionUtil.isNotEmpty(jobNodeConfs)){
//                    JobNodeConfig jobNodeConfig = iBdpVersionService.cronConfig(jobNodeConfs);
//                    if(null != jobNodeConfig){
//                        jobNodeConfig.setJobNodeId(workMenu.getId().intValue());
//                        jobNodeConfig.setJobType(1);
//                        List<JobNodeConf> jobNodeConfOlders = JobNodeConfig.jobNodeConfigTolist(jobNodeConfig);
//                        jobNodeConfMapper.upsertKVByJobNodeId(jobNodeConfOlders);
//                    }
//                }
//                log.error("系统参数重新刷新shell异常！error:" + e.getMessage());
//                throw new RRException("系统参数重新刷新shell异常！error:" + e.getMessage());
//            }
//        }

        // 虚拟化一个开始节点和一个结束节点，并且保存到hdfs上
        saveStartNodeAndEndNode(jobInfo.getProcId().longValue(), jobInfo.getId(), WorkFlowType.WORKFLOW.getCode());

        //更新连线和依赖
        changeJobNodeLayConfig(jobLinkAndJobNodeDto.getNodes(), newInsertDataList, workMenuJobRels.get(0).getWorkFlowId());

        //生成对应的任务工作流
        try {
            //generatedAutoWorkFlowXml(insertDataList, nodeIds, jobLinkAndJobNodeDto.getJobNodeConfig(), workMenuJobRels.get(0).getWorkFlowId(), jobLinkAndJobNodeDto.getProcId());
            generatedAutoWorkFlowXmlAndRoute(PojoUtils.listConvert(JobLink.class,insertDataList), nodeIds, jobLinkAndJobNodeDto.getJobNodeConfig(), workMenuJobRels.get(0).getWorkFlowId(), jobLinkAndJobNodeDto.getProcId(),nodeDtos);
        } catch (Exception e) {
            if(CollectionUtil.isNotEmpty(jobNodeConfs)){
                JobNodeConfig jobNodeConfig = iBdpVersionService.cronConfig(jobNodeConfs);
                if(null != jobNodeConfig){
                    jobNodeConfig.setJobNodeId(workMenu.getId().intValue());
                    jobNodeConfig.setJobType(1);
                    List<JobNodeConf> jobNodeConfOlders = JobNodeConfig.jobNodeConfigTolist(jobNodeConfig);
                    jobNodeConfMapper.upsertKVByJobNodeId(jobNodeConfOlders);
                }
            }
            log.error("WorkFlowXml"+insertDataList+"/"+nodeIds+"/"+jobLinkAndJobNodeDto.getJobNodeConfig()+"/"+workMenuJobRels.get(0).getWorkFlowId()+"/"+jobLinkAndJobNodeDto.getProcId());
            log.error("生成对应的定时调度的工作流异常！error:" + e.getMessage());
            throw new RRException("生成对应的定时调度的工作流异常！error:" + e.getMessage());
        }

        //保存节点和连线信息
        if (CollectionUtil.isNotEmpty(jobLinkAndJobNodeDto.getNodes())) {
            //更新节点的位置
            jobNodeInfoMapper.batchUpdate(jobLinkAndJobNodeDto.getNodes());
            //其它节点的位置给清除掉(相当于重画布中移除)
            jobNodeInfoMapper.batchUpdatePosOthers(jobLinkAndJobNodeDto.getNodes(), workMenuJobRels.get(0).getWorkFlowId());
        } else {
            jobNodeInfoMapper.clearPosOthers(workMenuJobRels.get(0).getWorkFlowId());
        }

        //更新连线
        jobLinkMapper.delete(new QueryWrapper<JobLink>().eq("linke_type", 0).eq("menu_id", jobLinkAndJobNodeDto.getWorkFlowMenuId()));

        if (CollectionUtil.isNotEmpty(newInsertDataList)) {
            jobLinkMapper.batchInsert(newInsertDataList);
        }

    }

    private void cheackJobStatusAndOozieId(OozieClient wc, String lastCorOozieJobId) {
        String status = null;
        try {
            status = wc.getStatus(lastCorOozieJobId);
        } catch (OozieClientException e) {
            e.printStackTrace();
            log.error("获取oozie的状态异常！error:{}",e.getMessage());
        }
        if("RUNNING".equalsIgnoreCase(status)){
            log.warn("停止任务的状态异常{}",lastCorOozieJobId);
            throw new RRException("oozie繁忙，请稍后尝试");
        }
    }

    private void refreshNodeShell(List<Map<String, Object>> jobNodeConfigs, Set<Integer> nodeIds, Integer isImport) throws Exception {
        if(CollectionUtil.isNotEmpty(jobNodeConfigs)){
            if(CollectionUtil.isNotEmpty(nodeIds)){
                for (Integer nodeId : nodeIds) {
                    JobNodeConf jobNodeConf = null;
                    JobNodeInfo jobNodeInfo = jobNodeInfoMapper.selectById(nodeId);
                    if(null == jobNodeInfo){
                        throw new RRException("节点id："+nodeId+"不存在");
                    }
                    if(Objects.equals(JobType.INPUT.getCode(),jobNodeInfo.getTypeId())|| Objects.equals(JobType.OUTPUT.getCode(),jobNodeInfo.getTypeId())){
                        jobNodeConf = jobNodeConfMapper.selectOne(new QueryWrapper<JobNodeConf>()
                                .eq("`key`","input_input_content")
                                .eq("job_node_id",nodeId).eq("job_type",0));
                    }
                    if(Objects.equals(JobType.SQL.getCode(),jobNodeInfo.getTypeId())){
                        jobNodeConf = jobNodeConfMapper.selectOne(new QueryWrapper<JobNodeConf>()
                                .eq("`key`","sql_statment")
                                .eq("job_node_id",nodeId).eq("job_type",0));
                    }
                    if(null != jobNodeConf){
                        String sql = jobNodeConf.getValue();
                        if(StringUtils.isNotEmpty(sql)){
                            if(isHaveSysparam(jobNodeConfigs,sql)){
                                List<JobNodeConf> jobNodeList = jobNodeConfMapper.selectList(new QueryWrapper<JobNodeConf>().eq("job_node_id",nodeId).eq("job_type",0));
                                if(CollectionUtil.isNotEmpty(jobNodeList)){
                                    JobInfo jobInfo = jobInfoMapper.selectById(jobNodeInfo.getJobId());
                                    if(null == jobInfo){
                                        throw new RRException("工作流id："+jobNodeInfo.getJobId()+"不存在");
                                    }
                                    jobNodeInfo.setProcId(jobInfo.getProcId());
                                    JobNodeConfig jobNodeConfig = iBdpVersionService.listJobNodeConfigList(jobNodeList,0);
                                    jobNodeInfo.setJobNodeConfig(jobNodeConfig);
                                    jobNodeInfo.setIsImport(isImport);
                                    jobNodeInfoService.saveJobNodeConfig(jobNodeInfo);
                                }
                            }
                        }
                    }
                }
            }
        }
    }

    @Override
    public Boolean isHaveSysparam(List<Map<String, Object>> jobNodeConfigs, String sql) {
        if(CollectionUtil.isNotEmpty(jobNodeConfigs) && StringUtils.isNotEmpty(sql)){
            for (Map<String, Object> jobNodeConfigMap : jobNodeConfigs) {
                if(null != jobNodeConfigMap){
                    Set<String> strings = jobNodeConfigMap.keySet();
                    for (String string : strings) {
                        String repStr = "${"+string+"}";
                        if(sql.contains(repStr)){
                            return Boolean.TRUE;
                        }
                    }
                }
            }
        }
        return Boolean.FALSE;
    }

    private void cheackExpressionNumber(JobNodeConfig jobNodeConfig) {
        if(null != jobNodeConfig && null != jobNodeConfig.getSchedule_type() && jobNodeConfig.getSchedule_type().equals(ScheduleType.REPEAT.getCode())){
            if(Objects.equals(2,jobNodeConfig.getCronType())){
                //cron表达式
                if(StringUtils.isEmpty(jobNodeConfig.getCronUrl())){
                    throw new RRException("cron表达式未配置！");
                }
                //验证cron表达式,最近10次的调度时间拿到，并且判断每两个时间的间隔时间>20min
                checkCron(jobNodeConfig.getCronUrl());
            }else{
                if(StringUtils.isEmpty(jobNodeConfig.getExpression())){
                    throw new RRException("请填写调度配置后再保存");
                }
                if("null".equals(jobNodeConfig.getExpression())){
                    throw new RRException("请填写调度配置后再保存");
                }
                JobConfUtil.resolveCrontab(jobNodeConfig);
            }
            if(null != jobNodeConfig && null != jobNodeConfig.getIsDurTime() && Objects.equals(1,jobNodeConfig.getIsDurTime())){
                String startTime = jobNodeConfig.getStartTime();
                String endTime = jobNodeConfig.getEndTime();
                if(StringUtils.isEmpty(startTime) || StringUtils.isEmpty(endTime)){
                    throw new RRException("调度的时间范围未配置！");
                }
                SimpleDateFormat simpleDateFormat1 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                Date parseStart = null;
                Date parseEnd = null;
                if(StringUtils.isNotEmpty(startTime)){
                    try {
                        parseStart = simpleDateFormat1.parse(startTime);
                    }catch (Exception e){
                        throw new RRException("调度的开始时间格式不正确");
                    }
                }else{
                    throw new RRException("调度的开始时间未配置");
                }
                if(StringUtils.isNotEmpty(endTime)){
                    try {
                        parseEnd = simpleDateFormat1.parse(endTime);
                    }catch (Exception e){
                        throw new RRException("调度的结束时间格式不正确");
                    }
                }else{
                    throw new RRException("调度的结束时间未配置");
                }
                //验证结束时间必须大于开始时间
                if(parseStart.after(parseEnd)){
                    throw new RRException("调度的结束时间必须大于开始时间");
                }
                //验证结束时间必须大于当前时间
                if(new Date().after(parseEnd)){
                    throw new RRException("调度的结束时间必须大于当前时间");
                }
            }
        }
        //验证对应的调度间隔时间段和出错次数
        if(null != jobNodeConfig && null != jobNodeConfig.getIsRetry() && Objects.equals(1,jobNodeConfig.getIsRetry())){
            Integer reTryInterval = jobNodeConfig.getReTryInterval();
            if(reTryInterval <= 0){
                throw new RRException("出错的间隔时间必须大于0分钟");
            }
        }

        //验证运行参数--换数据格式了
        if(null != jobNodeConfig && CollectionUtil.isNotEmpty(jobNodeConfig.getNodeParam())){
            List<JSONObject> nodeParam = jobNodeConfig.getNodeParam();
            Set<String> newKeys = new HashSet<>();
            for (JSONObject jsonObject : nodeParam) {
                newKeys.add(jsonObject.getString("key"));
            }
            if(newKeys.size() != nodeParam.size()){
                throw new RRException("存在重名的运行参数，请修改！");
            }
        }

        //验证是否开启动态时间参数
        if(null != jobNodeConfig && Objects.equals(1,jobNodeConfig.getIsOpenTime())){
            if(null == jobNodeConfig.getSysDurTime()){
                throw new RRException("间隔时间为空,请修改！");
            }
            if(StringUtils.isEmpty(jobNodeConfig.getSysStartTime())){
                throw new RRException("动态数据开始时间为空,请修改！");
            }
            if(StringUtils.isEmpty(jobNodeConfig.getSysEndTime())){
                throw new RRException("动态数据结束时间为空,请修改！");
            }
        }

        //验证对应的运行参数
        if(null != jobNodeConfig && CollectionUtil.isNotEmpty(jobNodeConfig.getSysParam())){
            List<Map<String, Object>> sysParam = jobNodeConfig.getSysParam();
            Set<String> newKeys = new HashSet<>();
            for (Map<String, Object> map : sysParam) {
                newKeys.addAll(map.keySet());
            }
            if(newKeys.size() != sysParam.size()){
                throw new RRException("存在重名的运行参数，请修改！");
            }
        }
    }

    private void checkCron(String cronUrl) {
        if (StringUtils.isEmpty(cronUrl)) {
            throw new RRException("crontab表达式不能为空！");
        }
        //必须是5位
        String tmpCronUrl = cronUrl.trim();
        String[] split = tmpCronUrl.split("\\s+");
        if(null == split || split.length != 5){
            throw new RRException("crontab表达式必须是5位！");
        }
        //crontab 少了一个秒 只有5位
        List<Date> dates = scheduleTimeService.cronTabNextDateList(new Date(), tmpCronUrl);
        if(CollectionUtil.isEmpty(dates)){
            throw new RRException("cron表达式对应的时间不能为空！");
        }
        for(int i = 1;i<10;i++){
            Date nextDate = dates.get(i);
            Date currentDate = dates.get(i - 1);
            if(nextDate.getTime() - currentDate.getTime() <20*60*1000){
                throw new RRException("暂时不支持调度间隔小于20分钟的自定义调度！");
            }
        }
    }

    private void deleteHdfs(Integer procId, Integer jobId) {
        String startDst = "workflow-app/" + WorkFlowType.WORKFLOW.getCode() + "/" + procId + "/" + jobId.toString() + "/workflow.xml";
        String corStartDst = "coordinator-app/" + WorkFlowType.WORKFLOW.getCode() + "/" + procId + "/" + jobId.toString() + "/coordinator.xml";
        HdfsUtil hdfsUtil = null;
        try {
            hdfsUtil = new HdfsUtil(bdpJobConfig);
            hdfsUtil.delete(startDst);
            hdfsUtil.delete(corStartDst);
        }catch (Exception e) {
            log.error("清空任务节点失败！error:{}", e.getMessage());
        }finally {
            if(null != hdfsUtil){
                hdfsUtil.close();
            }
        }
    }

    private void cheackJobNodeIsWorkFlowMenu(List<JobNodeInfoDto> nodes, Long menuId) {
        if (CollectionUtil.isEmpty(nodes)) {
            return;
        }
        List<Integer> workNodeIds = new ArrayList<>();
        List<WorkMenuJobRel> workMenuJobRels = workMenuJobRelMapper.selectList(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 0).eq("work_menu_id", menuId));
        if (CollectionUtil.isNotEmpty(workMenuJobRels)) {
            for (WorkMenuJobRel workMenuJobRel : workMenuJobRels) {
                workNodeIds.add(workMenuJobRel.getWorkFlowId().intValue());
            }
        }
        for (JobNodeInfoDto node : nodes) {
            if (!workNodeIds.contains(node.getId())) {
                throw new RRException("该工作流不能使用该任务节点：" + node.getName() + node.getId() + "！");
            }
        }
    }

    private void cheackWorkFlowJobLay(Long jobId, JobLinkAndJobNodeDto jobLinkAndJobNodeDto) {
        //这个job对应的工作流依赖
        List<WorkMenuJobRel> workMenuJobRels = workMenuJobRelMapper.selectList(new QueryWrapper<WorkMenuJobRel>().eq("work_flow_id", jobId).eq("work_type", 2));
        if (CollectionUtil.isNotEmpty(workMenuJobRels)) {
            //查询对应的菜单的运行状态
            WorkMenuJobRel workMenuJobRel = workMenuJobRels.get(0);
            WorkFlowDep workFlowDep = workFlowDepMapper.selectOne(new QueryWrapper<WorkFlowDep>().eq("work_menu_id", workMenuJobRel.getWorkMenuId()));
            if (null != workFlowDep) {
                if (workFlowDep.getJobLayStatus() == 1) {
                    //启用状态
                    if (null != workFlowDep.getLastRunState() && LastRunState.RUNNING.getCode() == workFlowDep.getLastRunState()) {
                        throw new RRException("工作流对应的工作流依赖正在运行中，请稍后重试！");
                    }
                    //启用状态，不能配置正常调度方式
                    JobNodeConfig jobNodeConfig = jobLinkAndJobNodeDto.getJobNodeConfig();
                    if (null != jobNodeConfig) {
                        if (Objects.equals(1, jobNodeConfig.getSchedule_type())) {
                            throw new RRException("工作流已关联启用的工作流依赖，不能配置正常调度方式！");
                        }
                    }
                }
                //验证自己的运行状态
                cheackJobStatus(jobId);
            }
        } else {
            //验证自己的运行状态
            cheackJobStatus(jobId);
        }
    }

    private void cheackJobStatus(Long jobId) {
        JobInfo jobInfo = jobInfoMapper.selectById(jobId);
        if (null != jobInfo) {
            if (null != jobInfo.getLastRunState() && LastRunState.RUNNING.getCode() == jobInfo.getLastRunState()) {
                throw new RRException("工作流正在运行中，请稍后重试！");
            }
        }
    }

    private void cheackJobNodeIsConfig(List<JobNodeInfoDto> nodes) {
        if (CollectionUtil.isEmpty(nodes)) {
            throw new RRException("请配置节点连线！");
        }
        //验证对应的节点是否都配置有对应的配置
        for (JobNodeInfoDto node : nodes) {
            List<JobNodeConf> jobNodeConfs = jobNodeConfMapper.selectList(new QueryWrapper<JobNodeConf>().eq("job_node_id", node.getId()).eq("job_type", 0));
            if (CollectionUtil.isEmpty(jobNodeConfs)) {
                throw new RRException("节点" + node.getName() + "_" + node.getId() + "未配置");
            }
        }
    }


    private void generatedAutoWorkFlowXmlAndRoute(List<JobLink> insertDataList, Set<Integer> nodeIds, JobNodeConfig jobNodeConfig, Long jobId, Integer procId, Set<JobNodeInfoDto> nodeDtos) throws Exception {
        JobInfo jobInfo = jobInfoMapper.selectById(jobId.intValue());
        if (null == jobInfo) {
            throw new RRException("工作流不存在!");
        }
        //连线
        if(CollectionUtil.isEmpty(nodeIds)){
            throw new RRException("工作流未配置节点!");
        }
        List<JobNodeInfoDto> jobNodeInfoDtos = jobNodeInfoMapper.selectListByIds1(nodeIds);
        if(CollectionUtil.isEmpty(jobNodeInfoDtos)){
            throw new RRException("工作流未配置节点!");
        }

        Set<Integer> haveNullNodeIds = new HashSet<>();
        for (JobNodeInfoDto jobNodeInfoDto : jobNodeInfoDtos) {
            //节点是否空跑
            JobNodeInfoDto jobNodeInfoDto1 = nodeDtos.stream().filter(o -> Objects.equals(o.getId(), jobNodeInfoDto.getId())).findAny().orElse(null);
            Integer isNullRun = 0;
            if(null != jobNodeInfoDto1){
                isNullRun = jobNodeInfoDto1.getIsNullRun();
                if(Objects.equals(1,jobNodeInfoDto1.getIsNullRun())){
                    haveNullNodeIds.add(jobNodeInfoDto.getId());
                }
            }
            jobNodeInfoDto.setIsNullRun(isNullRun);

            //路由信息
            List<JobNodeLayConfigDto> jobNodeLayConfigDtos = jobNodeLayConfigMapper.selectDtoListByJobNodeId(jobNodeInfoDto.getId(), jobNodeInfoDto.getJobId());
            jobNodeInfoDto.setJobNodeLayConfigDtoList(jobNodeLayConfigDtos);
            jobNodeInfoDto.setWorkFlowNodeName(JobType.ObjOf(jobNodeInfoDto.getTypeId()).getEnName() + "_" +  jobNodeInfoDto.getId());
            Boolean haveCodition = cheackSubNodeIsLayCondition(jobNodeInfoDto);
            if(haveCodition){
                jobNodeInfoDto.setHaveCondition(1);
            }
            //保存配置的失败路由的上游节点和其它的上游节点
            Set<Integer> errorUpNodes = new HashSet<>();
            Set<Integer> otherUpNodes = new HashSet<>();
            Boolean haveError = cheackSubNodeIsErrorCondition(jobNodeInfoDto,errorUpNodes,otherUpNodes);
            if(haveError){
                jobNodeInfoDto.setHaveError(1);
                if(CollectionUtil.isNotEmpty(errorUpNodes)){
                    jobNodeInfoDto.setErrorUpNodes(errorUpNodes);
                }
                if(CollectionUtil.isNotEmpty(otherUpNodes)){
                    jobNodeInfoDto.setOtherUpNodes(otherUpNodes);
                }
            }
        }

        //生成workflow.xml
        //根据连线和路由条件，得到对应的workflow,并且保存到hdfs上
        OozieClient wc = new OozieClient(oozieConfig.getUrl());
        HdfsUtil hdfsUtil = new HdfsUtil(bdpJobConfig);
        JobUtil jobUtil = new JobUtil(hdfsUtil, wc);

        //组装数据和虚拟节点并且上传
        List<List<JobNodeInfoDto>> allLinkedDtoList = packRouteAndVirlNode(jobNodeInfoDtos,insertDataList,jobInfo,hdfsUtil);

        Properties properties = null;
        try {
            properties = jobUtil.createOrUpdateHaveConditionJob(procId, getQueueName(procId), jobId.intValue(), allLinkedDtoList, false, true,
                    haveNullNodeIds, ReRunTypeEnum.RE_ALL_NODE.getId(),jobNodeConfig);
        } catch (Exception e) {
            if(null != hdfsUtil){
                hdfsUtil.close();
            }
            log.error("生成workFlowJOb异常，error:" + e.getMessage());
            throw new RRException("生成workFlowJOb异常!");
        }

        //如果是需要定时调度，则先暂停之前的定时调度，然后重新生成对应的定时调度
        //如果有调度，则暂停之前的调度
        if (StringUtils.isNotEmpty(jobInfo.getLastCorOozieJobId())) {
            jobUtil.stopJob(jobInfo.getLastCorOozieJobId());
            cheackJobStatusAndOozieId(wc,jobInfo.getLastCorOozieJobId());
            ExecutorService executorService = Executors.newSingleThreadExecutor();
            executorService.execute(()->{
                //暂停调度的工作流
                jobInfoService.stopAutoJob(jobInfo.getId());
            });
        }


        if (null == jobNodeConfig || null == jobNodeConfig.getSchedule_type() || jobNodeConfig.getSchedule_type() != 1) {
            if(null != hdfsUtil){
                hdfsUtil.close();
            }
            return;
        }
        try {
            properties = jobUtil.createOrUpdateHaveConditionJob(procId,getQueueName(procId), jobId.intValue(), allLinkedDtoList, true, false,haveNullNodeIds, ReRunTypeEnum.RE_ALL_NODE.getId(),jobNodeConfig);
        } catch (Exception e) {
            if(null != hdfsUtil){
                hdfsUtil.close();
            }
            log.error("生成workFlowJOb异常，error:" + e.getMessage());
            throw new RRException("生成定时workFlowJOb异常!");
        }

        JSONObject objStartTime = new JSONObject();
        String oozieJobId = jobUtil.runCoordinatorJob(procId, jobId.intValue(), properties,
                jobNodeConfig.getExpression(),
                jobNodeConfig.getScheduleOrInterval(),
                jobNodeConfig.getIsDurTime(),
                jobNodeConfig.getStartTime(),
                jobNodeConfig.getEndTime(),
                jobNodeConfig.getIsRetry(),
                jobNodeConfig.getReTryMax(),
                jobNodeConfig.getReTryInterval(),objStartTime,jobNodeConfig.getCronType(),jobNodeConfig.getCronUrl());

        //追加一个开始节点，让其自动回调给rest_api的start。获取到对应的任务ozzieId,方便轮询查询对应的任务的日志信息
        //jobName : type/procId/jobId
        String beginShell = pingRestApiStartShell(oozieJobId, WorkFlowType.WORKFLOW.getCode() + "/" + procId + "/" + jobId,WorkFlowType.WORKFLOW.getCode());
        String startDst = "workflow-app/" + WorkFlowType.WORKFLOW.getCode() + "/" + procId + "/" + jobId.toString() + "/StartTwo_0.sh";
        hdfsUtil.writeFile(beginShell.getBytes(), startDst);
        if(null != hdfsUtil){
            hdfsUtil.close();
        }
        jobInfo.setLastCorOozieJobId(oozieJobId);
        jobInfo.setLastRunNum(1);
        String startTime = objStartTime.getString("startTime");
        Date date01 = new Date();
        try {
            SimpleDateFormat simpleDateFormat1 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            date01 = simpleDateFormat1.parse(startTime);
        }catch (Exception e){
            log.error("转换时间异常！error:{}",e.getMessage());
        }
        jobInfo.setCronLastRunTime(date01);
        if (Objects.equals(LastRunState.RUNNING.getCode(), jobInfo.getLastRunState())) {
            jobInfo.setLastRunState(LastRunState.STOP.getCode());
        }
        jobInfoMapper.updateById(jobInfo);

    }

    private Boolean cheackSubNodeIsErrorCondition(JobNodeInfoDto jobNodeInfoDto, Set<Integer> errorUpNodes, Set<Integer> upNodes) {
        Boolean falge = Boolean.FALSE;
        if(null == jobNodeInfoDto){
            return falge;
        }
        List<JobNodeLayConfigDto> jobNodeLayConfigDtoList = jobNodeInfoDto.getJobNodeLayConfigDtoList();
        if(CollectionUtil.isNotEmpty(jobNodeLayConfigDtoList)){
            for (JobNodeLayConfigDto jobNodeLayConfigDto : jobNodeLayConfigDtoList) {
                if( Objects.equals(RouteTypeEnum.STATUS.getId(),jobNodeLayConfigDto.getConfigType()) && Objects.equals(jobNodeLayConfigDto.getConditionValue(),"1")){
                    falge = Boolean.TRUE;
                    errorUpNodes.add(jobNodeLayConfigDto.getUpNodeId());
                    continue;
                }
                upNodes.add(jobNodeLayConfigDto.getUpNodeId());
            }
        }
        return falge;
    }

    private String getQueueName(Integer procId) {
        ProcInfo procInfo = procInfoMapper.selectById(procId);
        if(null == procInfo){
            return "default";
        }

        if(null != procInfo.getQueueId()){
            RealtimeQueueEntity realtimeQueueEntity = realtimeQueueMapper.selectById(procInfo.getQueueId());
            if(null != realtimeQueueEntity){
                return realtimeQueueEntity.getQueueName();
            }
        }
        return "default";
    }

    private List<List<JobNodeInfoDto>> packRouteAndVirlNode(List<JobNodeInfoDto> jobNodeInfoDtos, List<JobLink> insertDataList, JobInfo jobInfo, HdfsUtil hdfsUtil) {
        List<JobLink> newJobLinks = new ArrayList<>();
        List<JSONObject> addLinks = new ArrayList<>();
        List<JobNodeInfoDto> newJobNodeInfoDtos = new ArrayList<>(jobNodeInfoDtos);
        //判断是否需要虚拟节点
        Set<JobNodeInfoDto> virNullNodeList = new HashSet<>();
        for (JobNodeInfoDto jobNodeInfoDto : jobNodeInfoDtos) {
            virNullNodeList.add(jobNodeInfoDto);
        }
        if(CollectionUtil.isNotEmpty(virNullNodeList)){
            upShellVir(virNullNodeList,3,hdfsUtil,jobInfo.getProcId(),jobInfo.getId());
        }
        if(CollectionUtil.isNotEmpty(insertDataList)){
            for (JobNodeInfoDto jobNodeInfoDto : jobNodeInfoDtos) {
                List<JobLink> dstJobLinkList = insertDataList.stream().filter(jobLink -> Objects.equals(jobLink.getSrcNodeId(), jobNodeInfoDto.getId())).collect(Collectors.toList());
                if(CollectionUtil.isNotEmpty(dstJobLinkList)){
                    List<JobLink> unConfigList = new ArrayList<>();
                    //挑选出有路由条件的集合数据
                    List<JobLink> filtConditionList = filtCondition(jobNodeInfoDtos,dstJobLinkList,unConfigList);
                    if(CollectionUtil.isNotEmpty(filtConditionList)){
                        if(filtConditionList.size() == dstJobLinkList.size()){
                            //全部都是路由条件，放入默认的虚拟节点
                            newJobLinks.addAll(dstJobLinkList);
                            addLinks.add(pingDefNode(jobNodeInfoDto.getId()));
                            newJobNodeInfoDtos.add(pingDefJobNodeInfoDto(jobNodeInfoDto.getId()));
                        } else {
                            //有部分是路由条件的，有部分不是路由条件的
                            newJobLinks.addAll(unConfigList);
                            newJobNodeInfoDtos.add(pingVirlJobNodeInfoDto(jobNodeInfoDto.getId()));
                            addLinks.add(pingVirlNode(jobNodeInfoDto.getId()));
                            //虚拟的聚合节点
                            newJobNodeInfoDtos.add(pingVirlJobNodeInfoEnd(jobNodeInfoDto.getId()));
                            for (JobLink jobLink : filtConditionList) {
                                addLinks.add(pingVirlNodeStart(jobNodeInfoDto.getId(),jobLink.getDstNodeId()));
                                addLinks.add(pingVirlJobNodeEnd(jobLink.getDstNodeId(),jobNodeInfoDto.getId()));
                            }
                            newJobNodeInfoDtos.add(pingDefJobNodeInfoDto(jobNodeInfoDto.getId()));
                            addLinks.add(pingDefNodeByVirl(jobNodeInfoDto.getId()));
                            //虚拟的结束连接默认聚合节点
                            addLinks.add(pingDefNodeEnd(jobNodeInfoDto.getId()));
                        }
                    } else{
                        newJobLinks.addAll(dstJobLinkList);
                    }
                }
            }

        }

        //虚拟开始节点和结束节点以及连线
        List<List<JobNodeInfoDto>> allLinkedDtoList = listAllJobNodeDtoList(newJobLinks,addLinks,newJobNodeInfoDtos);

        //上传workflow.xml文件和虚拟的节点和获取对应的config
        Set<JobNodeInfoDto> deftNodeList = new HashSet<>();
        Set<JobNodeInfoDto> virNodeList = new HashSet<>();
        Set<JobNodeInfoDto> virNodeEndList = new HashSet<>();
        for (List<JobNodeInfoDto> nodeInfoDtos : allLinkedDtoList) {
            List<JobNodeInfoDto> deft = nodeInfoDtos.stream().filter(jobNodeInfoDto -> Objects.equals(jobNodeInfoDto.getVirType(), 1)).collect(Collectors.toList());
            if(CollectionUtil.isNotEmpty(deft)){
                deftNodeList.addAll(deft);
            }
            List<JobNodeInfoDto> virNode = nodeInfoDtos.stream().filter(jobNodeInfoDto -> Objects.equals(jobNodeInfoDto.getVirType(), 2)).collect(Collectors.toList());
            if(CollectionUtil.isNotEmpty(virNode)){
                virNodeList.addAll(virNode);
            }
            List<JobNodeInfoDto> virNodeEnd = nodeInfoDtos.stream().filter(jobNodeInfoDto -> Objects.equals(jobNodeInfoDto.getVirType(), 3)).collect(Collectors.toList());
            if(CollectionUtil.isNotEmpty(virNodeEnd)){
                virNodeEndList.addAll(virNodeEnd);
            }
        }
        if(CollectionUtil.isNotEmpty(deftNodeList)){
            upShellVir(deftNodeList,1,hdfsUtil,jobInfo.getProcId(),jobInfo.getId());
        }
        if(CollectionUtil.isNotEmpty(virNodeList)){
            upShellVir(virNodeList,2,hdfsUtil,jobInfo.getProcId(),jobInfo.getId());
        }
        if(CollectionUtil.isNotEmpty(virNodeEndList)){
            upShellVir(virNodeEndList,4,hdfsUtil,jobInfo.getProcId(),jobInfo.getId());
        }

        return allLinkedDtoList;
    }

    private JSONObject pingDefNodeEnd(Integer id) {
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("dstNodeId",VirtualTypeEnum.VIRTUAL_SUM_END.getName() + "_" + id);
        jsonObject.put("srcNodeId",VirtualTypeEnum.VIRTUAL_DEF.getName() + "_" + id);
        return jsonObject;
    }

    private JSONObject pingVirlJobNodeEnd(Integer dstNodeId, Integer id) {
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("srcNodeId",dstNodeId);
        jsonObject.put("dstNodeId", VirtualTypeEnum.VIRTUAL_SUM_END.getName() + "_" + id);
        return jsonObject;
    }

    private JobNodeInfoDto pingVirlJobNodeInfoEnd(Integer jobNodeInfoDtoId) {
        JobNodeInfoDto jobNodeInfoDto = new JobNodeInfoDto();
        jobNodeInfoDto.setWorkFlowNodeName(VirtualTypeEnum.VIRTUAL_SUM_END.getName() + "_" + jobNodeInfoDtoId);
        jobNodeInfoDto.setVirType(VirtualTypeEnum.VIRTUAL_SUM_END.getId());
        return jobNodeInfoDto;
    }

//    /***
//     * 生成对应的workFlow让其执行
//     * @param edges  节点连线
//     * @param jobNodeConfig  作业流的配置属性
//     * @param jobId          任务id
//     */
//    private void generatedAutoWorkFlowXml(List<JobLinkDto> edges, Set<Integer> nodeIds, JobNodeConfig jobNodeConfig, Long jobId, Integer procId) throws Exception {
//        //定为：保存则生成workflow,后面直接使用这个workflow.(统一入口)---增加路由功能，则统一不了
//        //在保存节点配置信息的地方
//        JobInfo jobInfo = jobInfoMapper.selectById(jobId.intValue());
//        if (null == jobInfo) {
//            throw new RRException("工作流不存在!");
//        }
//        //获取到节点的所有的连线信息
//        List<List<JobNodeInfoDto>> allLinkedDtoList = listAllJobNodeList(edges, nodeIds);
//
//        //TODO:hujz 查询出对应的节点配置的路由条件
//
//        //根据连线和路由条件，得到对应的workflow,并且保存到hdfs上
//        OozieClient wc = new OozieClient(oozieConfig.getUrl());
//        HdfsUtil hdfsUtil = new HdfsUtil(bdpJobConfig);
//        JobUtil jobUtil = new JobUtil(hdfsUtil, wc);
//        Properties conf = null;
//        try {
//            conf = jobUtil.createOrUpdateJob(procId, jobId.intValue(), allLinkedDtoList, false, true);
//        } catch (Exception e) {
//            log.error("生成workFlowJOb异常，error:" + e.getMessage());
//            throw new RRException("生成workFlowJOb异常!");
//        }
//        if (null == jobNodeConfig || null == jobNodeConfig.getSchedule_type() || jobNodeConfig.getSchedule_type() != 1) {
//            return;
//        }
//
//        try {
//            conf = jobUtil.createOrUpdateJob(procId, jobId.intValue(), allLinkedDtoList, true, true);
//        } catch (Exception e) {
//            log.error("生成workFlowJOb异常，error:" + e.getMessage());
//            throw new RRException("生成定时workFlowJOb异常!");
//        }
//
//        //如果有调度，则暂停之前的调度
//        if (StringUtils.isNotEmpty(jobInfo.getLastCorOozieJobId())) {
//            jobUtil.stopJob(jobInfo.getLastCorOozieJobId());
//        }
//
//        String oozieJobId = jobUtil.runCoordinatorJob(procId, jobId.intValue(), conf, jobNodeConfig.getExpression(), jobNodeConfig.getScheduleOrInterval().toString(),
//                jobNodeConfig.getIsDurTime(),
//                jobNodeConfig.getStartTime(),
//                jobNodeConfig.getEndTime(),
//                jobNodeConfig.getIsRetry(),
//                jobNodeConfig.getReTryMax(),
//                jobNodeConfig.getReTryInterval());
//
//        //追加一个开始节点，让其自动回调给rest_api的start。获取到对应的任务ozzieId,方便轮询查询对应的任务的日志信息
//        //jobName : type/procId/jobId
//        String beginShell = pingRestApiStartShell(oozieJobId, WorkFlowType.WORKFLOW.getCode() + "/" + procId + "/" + jobId,WorkFlowType.WORKFLOW.getCode());
//        String startDst = "workflow-app/" + WorkFlowType.WORKFLOW.getCode() + "/" + procId + "/" + jobId.toString() + "/StartTwo_0.sh";
//        hdfsUtil.writeFile(beginShell.getBytes(), startDst);
//
//        //保存对应的任务的历史
////        List<JobNodeInfo> jobNodeInfos = jobNodeInfoMapper.selectList(new QueryWrapper<JobNodeInfo>().eq("job_id", jobId));
////        jobNodeInfoService.initJobHistory(edges, PojoUtils.listConvert(JobNodeInfoDto.class, jobNodeInfos), "自动运行", jobId.intValue(), oozieJobId);
//
//        jobInfo.setLastCorOozieJobId(oozieJobId);
//        jobInfo.setLastRunNum(1);
//        jobInfoMapper.updateById(jobInfo);
//    }

    @Override
    public void runWorkJob(Long jobMenuId){
        //运行工作流
        //当前任务运行状态；自己创建的人，自己可以运行；生成对应的workflow,并且运行；回调对应的日志信息
        WorkMenu workMenu = workMenuMapper.selectById(jobMenuId);
        if (null == workMenu) {
            throw new RRException("任务流菜单不存在!");
        }

        //验证是否有权限
        iWorkMenuService.cheackProcId(workMenu.getProcId(), ShiroUtils.getUserId(), true, workMenu.getCreatePer(),"user",null);

        WorkMenuJobRel workMenuJobRel = workMenuJobRelMapper.selectOne(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 1).eq("work_menu_id", jobMenuId));
        if (null == workMenuJobRel) {
            throw new RRException("作业流菜单和节点关联表数据不存在!");
        }
        JobInfo jobInfo = jobInfoMapper.selectById(workMenuJobRel.getWorkFlowId());
        if (null == jobInfo) {
            throw new RRException("任务流不存在!");
        }
        //运行状态
        if (null != jobInfo.getLastRunState() && LastRunState.RUNNING.getCode() == jobInfo.getLastRunState()) {
            throw new RRException("作业流正在运行中，请稍后重试!");
        }
        //验证引用了这个工作流的工作流依赖对应的工作流是否在运行中
        cheackWorkFlowStatus(jobInfo.getId());

        //查询其下所有的连线信息
        List<JobLink> jobLinks = jobLinkMapper.selectList(new QueryWrapper<JobLink>().eq("linke_type", 0).eq("menu_id", jobMenuId));
        //查询其下所有的节点
        List<JobNodeInfo> jobNodeInfos1 = jobNodeInfoMapper.selectList(new QueryWrapper<JobNodeInfo>().eq("job_id", jobInfo.getId()).isNotNull("pos_x").isNotNull("pos_y"));
        if (CollectionUtil.isEmpty(jobNodeInfos1)) {
            throw new RRException("作业流未配置节点!");
        }
        Set<Integer> nodeIds = new HashSet<>();
        for (JobNodeInfo jobNodeInfo : jobNodeInfos1) {
            nodeIds.add(jobNodeInfo.getId());
        }
        List<JobNodeInfoDto> jobNodeInfoDtos = jobNodeInfoMapper.selectListByIds(nodeIds);
        if(CollectionUtil.isEmpty(jobNodeInfoDtos)){
            throw new RRException("工作流未配置节点!");
        }

        Set<Integer> haveNullNodeIds = new HashSet<>();
        for (JobNodeInfoDto jobNodeInfoDto : jobNodeInfoDtos) {
            //节点是否空跑
            if(Objects.equals(1,jobNodeInfoDto.getIsNullRun())){
                haveNullNodeIds.add(jobNodeInfoDto.getId());
            }
            //路由信息
            List<JobNodeLayConfigDto> jobNodeLayConfigDtos = jobNodeLayConfigMapper.selectDtoListByJobNodeId(jobNodeInfoDto.getId(), jobNodeInfoDto.getJobId());
            jobNodeInfoDto.setJobNodeLayConfigDtoList(jobNodeLayConfigDtos);
            jobNodeInfoDto.setWorkFlowNodeName(JobType.ObjOf(jobNodeInfoDto.getTypeId()).getEnName() + "_" +  jobNodeInfoDto.getId());
            Boolean haveCodition = cheackSubNodeIsLayCondition(jobNodeInfoDto);
            if(haveCodition){
                jobNodeInfoDto.setHaveCondition(1);
            }
        }
        //获取到节点的所有的连线信息
        //生成workflow.xml
        //根据连线和路由条件，得到对应的workflow,并且保存到hdfs上
        OozieClient wc = new OozieClient(oozieConfig.getUrl());
        HdfsUtil hdfsUtil = null;
        try {
            hdfsUtil = new HdfsUtil(bdpJobConfig);
        } catch (Exception e) {
            log.error("生成hdfs帮助类异常，error:" + e.getMessage());
            e.printStackTrace();
            throw new RRException("生成hdfs帮助类异常!");
        }
        JobUtil jobUtil = new JobUtil(hdfsUtil, wc);

        //组装数据和虚拟节点并且上传
        List<List<JobNodeInfoDto>> allLinkedDtoList = packRouteAndVirlNode(jobNodeInfoDtos,jobLinks,jobInfo,hdfsUtil);

        //根据连线和路由条件，得到对应的workflow,并且保存到hdfs上
        String oozieJobId = "";
        try {
            Properties properties = jobUtil.createOrUpdateHaveConditionJob(jobInfo.getProcId(), getQueueName(jobInfo.getProcId()), jobInfo.getId(), allLinkedDtoList, false, false, haveNullNodeIds, ReRunTypeEnum.RE_ALL_NODE.getId(), null);
            oozieJobId = wc.run(properties);
            if(null != hdfsUtil){
                hdfsUtil.close();
            }
        } catch (Exception e) {
            if(null != hdfsUtil){
                hdfsUtil.close();
            }
            e.printStackTrace();
            log.error("生成workFlowJOb异常，error:" + e.getMessage());
            throw new RRException("生成定时workFlowJOb异常!");
        }finally {
            if(null != hdfsUtil){
                hdfsUtil.close();
            }
        }

        //保存对应的任务的历史
        jobInfo.setLastOozieJobId(oozieJobId);
        jobInfo.setLastRunState(LastRunState.RUNNING.getCode());
        jobInfo.setLastRunType(ShiroUtils.getUserId().intValue());
        jobInfoService.updateById(jobInfo);

        //初始化job的历史记录
        List<JobNodeInfo> jobNodeInfos = jobNodeInfoMapper.selectList(new QueryWrapper<JobNodeInfo>()
                .eq("job_id", jobInfo.getId()).isNotNull("pos_x").isNotNull("pos_y"));
        String cronStartTime = null,cronEndTime = null,cronQueueName = null;
        //查询到对应的任务配置的参数时间
        JobCronDto jobCronDto = getCronTypeByJobId(jobInfo.getId());
        if(null != jobCronDto){
            if(Objects.equals(1,jobCronDto.getIsOpenTime())){
                cronStartTime = jobCronDto.getSysStartTime();
                cronEndTime = jobCronDto.getSysEndTime();
            }
            cronQueueName = jobCronDto.getQueueName();
        }
        jobNodeInfoService.initJobHistory(PojoUtils.listConvert(JobLinkDto.class, jobLinks),
                PojoUtils.listConvert(JobNodeInfoDto.class, jobNodeInfos),
                "手动运行初始化数据",
                jobInfo.getId(), oozieJobId, ReRunTypeEnum.RE_ALL_NODE.getId(), 0,ShiroUtils.getUserId().intValue(),cronStartTime,cronEndTime,cronQueueName);


        //查询状态，让其进行回调，更新job任务的日志状态
        int count = 10;
        while (count > 0){
            ThreadUtil.sleep(100);
            try {
                String s = jobRestFeignApi.pollingWorkFlow(jobInfo.getId(), oozieJobId);
                if("success".equals(s)){
                    count = 0;
                }
            }catch (Exception e){
                log.error("请求工作流轮询异常！error:{}",e.getMessage());
            }
            count--;
        }


        //webscoket推送状态
        WorkFlowStatusVO workFlowStatusVO = new WorkFlowStatusVO();
        workFlowStatusVO.setId(Long.valueOf(jobInfo.getId()));
        workFlowStatusVO.setStatus(LastRunState.RUNNING.getCode());
        workFlowStatusVO.setOozieId(oozieJobId);
        //查询到其下都有的节点，都是未执行
        if (CollectionUtil.isNotEmpty(jobNodeInfoDtos)) {
            List<WorkFlowStatusVO> nodeList = new ArrayList<>();
            for (JobNodeInfo jobNodeInfo : jobNodeInfos) {
                WorkFlowStatusVO workFlowStatusVO1 = new WorkFlowStatusVO();
                workFlowStatusVO1.setId(Long.valueOf(jobNodeInfo.getId()));
                workFlowStatusVO1.setStatus(LastRunState.NOTRUN.getCode());
                workFlowStatusVO1.setOozieId(oozieJobId);
                nodeList.add(workFlowStatusVO1);
            }
            workFlowStatusVO.setNodeList(nodeList);
        }
        pushWebStatus(jobMenuId,workFlowStatusVO);


    }

    private void pushWebStatus(Long jobMenuId, WorkFlowStatusVO workFlowStatusVO) {
        //遍历注册到这个集合中的所有的需要查询的数据
        ConcurrentHashMap<String, Session> sessionMap = WorkFlowStatusSocketServer.listSessionMap();
        Iterator<Map.Entry<String, Session>> entries = sessionMap.entrySet().iterator();
        while (entries.hasNext()) {
            Map.Entry<String, Session> entry = entries.next();
            String[] split = entry.getKey().split("_");
            if(null != split && split.length >= 3){
                if(split[0].equals(jobMenuId.toString())){
                    if (null != workFlowStatusVO) {
                        WorkFlowStatusSocketServer.sendMessage(entry.getValue(), JSONObject.toJSONString(workFlowStatusVO));
                    }
                }
            }
        }
    }

    @Override
    public void runNodeJob(WorkMenuDto workMenuDto) {
        //运行当前节点
        if(null == workMenuDto.getId()){
            throw new RRException("节点id必须传递!");
        }
        JobNodeInfo jobNodeInfo = jobNodeInfoMapper.selectById(workMenuDto.getId());
        if(null == jobNodeInfo){
            throw new RRException("节点不存在!");
        }
        if(null == jobNodeInfo.getPosX() || null == jobNodeInfo.getPosY()){
            throw new RRException("节点未对应到工作流!");
        }
        JobInfo jobInfo = jobInfoMapper.selectById(jobNodeInfo.getJobId());
        if(null == jobInfo){
            throw new RRException("节点对应的工作流不存在!");
        }
        WorkMenuJobRel workMenuJobRel = workMenuJobRelMapper.selectOne(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 1).eq("work_flow_id", jobInfo.getId()));
        if(null == workMenuJobRel){
            throw new RRException("节点对应的工作流菜单不存在!");
        }

        //验证是否有权限
        iWorkMenuService.cheackProcId(jobInfo.getProcId(), ShiroUtils.getUserId(), true, jobInfo.getCreatePer(),"user",null);
        //运行状态
        if (null != jobInfo.getLastRunState() && LastRunState.RUNNING.getCode() == jobInfo.getLastRunState()) {
            throw new RRException("作业流正在运行中，请稍后重试!");
        }
        //验证引用了这个工作流的工作流依赖对应的工作流是否在运行中
        cheackWorkFlowStatus(jobInfo.getId());

        Set<Integer> haveNullNodeIds = new HashSet<>();
        //节点是否空跑
        if(Objects.equals(1,jobNodeInfo.getIsNullRun())){
            haveNullNodeIds.add(jobNodeInfo.getId());
        }
        List<JobNodeInfoDto> jobNodeInfoDtos = new ArrayList<>();
        JobNodeInfoDto jobNodeInfoDto = new JobNodeInfoDto();
        BeanUtils.copyProperties(jobNodeInfo,jobNodeInfoDto);
        jobNodeInfoDto.setWorkFlowNodeName(JobType.ObjOf(jobNodeInfoDto.getTypeId()).getEnName() + "_" +  jobNodeInfoDto.getId());
        jobNodeInfoDtos.add(jobNodeInfoDto);


        //获取到节点的所有的连线信息
        //生成workflow.xml
        //根据连线和路由条件，得到对应的workflow,并且保存到hdfs上
        OozieClient wc = new OozieClient(oozieConfig.getUrl());
        HdfsUtil hdfsUtil = null;
        try {
            hdfsUtil = new HdfsUtil(bdpJobConfig);
        } catch (Exception e) {
            log.error("生成hdfs帮助类异常，error:" + e.getMessage());
            throw new RRException("生成hdfs帮助类异常!");
        }
        JobUtil jobUtil = new JobUtil(hdfsUtil, wc);

        //组装数据和虚拟节点并且上传
        List<List<JobNodeInfoDto>> allLinkedDtoList = packRouteAndVirlNode(jobNodeInfoDtos,null,jobInfo,hdfsUtil);


        //根据连线和路由条件，得到对应的workflow,并且保存到hdfs上
        String oozieJobId = "";
        try {
            Properties properties = jobUtil.createOrUpdateHaveConditionJob(jobInfo.getProcId(), getQueueName(jobInfo.getProcId()), jobInfo.getId(), allLinkedDtoList, false, true, haveNullNodeIds,ReRunTypeEnum.RE_CURRENT_NODE.getId(), null);
            oozieJobId = wc.run(properties);
            if(null != hdfsUtil){
                hdfsUtil.close();
            }
        } catch (Exception e) {
            if(null != hdfsUtil){
                hdfsUtil.close();
            }
            log.error("生成workFlowJOb异常，error:" + e.getMessage());
            throw new RRException("生成workFlowJOb异常!");
        }finally {
            if(null != hdfsUtil){
                hdfsUtil.close();
            }
        }

        //保存对应的任务的历史
        jobInfo.setLastOozieJobId(oozieJobId);
        jobInfo.setLastRunState(LastRunState.RUNNING.getCode());
        jobInfo.setLastRunType(ShiroUtils.getUserId().intValue());
        jobInfoService.updateById(jobInfo);

        //查询其下所有的连线信息
        List<JobLink> jobLinks = jobLinkMapper.selectList(new QueryWrapper<JobLink>().eq("linke_type", 0).eq("menu_id", workMenuJobRel.getWorkMenuId()));
        //初始化job的历史记录
        List<JobNodeInfo> jobNodeInfos = jobNodeInfoMapper.selectList(new QueryWrapper<JobNodeInfo>()
                .eq("job_id", jobInfo.getId()).isNotNull("pos_x").isNotNull("pos_y"));
        String cronStartTime = null,cronEndTime = null,cronQueueName = null;
        //查询到对应的任务配置的参数时间
        JobCronDto jobCronDto = getCronTypeByJobId(jobInfo.getId());
        if(null != jobCronDto){
            if(Objects.equals(1,jobCronDto.getIsOpenTime())){
                cronStartTime = jobCronDto.getSysStartTime();
                cronEndTime = jobCronDto.getSysEndTime();
            }
            cronQueueName = jobCronDto.getQueueName();
        }
        jobNodeInfoService.initJobHistory(PojoUtils.listConvert(JobLinkDto.class, jobLinks), PojoUtils.listConvert(JobNodeInfoDto.class, jobNodeInfos), "手动运行初始化数据", jobInfo.getId(), oozieJobId,ReRunTypeEnum.RE_CURRENT_NODE.getId(),jobNodeInfo.getId(),ShiroUtils.getUserId().intValue(), cronStartTime, cronEndTime, cronQueueName);


        //查询状态，让其进行回调，更新job任务的日志状态
        int count = 10;
        while (count > 0){
            ThreadUtil.sleep(100);
            try {
                String s = jobRestFeignApi.pollingWorkFlow(jobInfo.getId(), oozieJobId);
                if("success".equals(s)){
                    count = 0;
                }
            }catch (Exception e){
                log.error("请求运行当前节点轮询异常！error:{}",e.getMessage());
            }
            count--;
        }
    }

    @Override
    public void runUpJob(WorkMenuDto workMenuDto) {
        //运行当前节点和上游节点
        if(null == workMenuDto.getId()){
            throw new RRException("节点id必须传递!");
        }
        JobNodeInfo jobNodeInfo = jobNodeInfoMapper.selectById(workMenuDto.getId());
        if(null == jobNodeInfo){
            throw new RRException("节点不存在!");
        }
        if(null == jobNodeInfo.getPosX() || null == jobNodeInfo.getPosY()){
            throw new RRException("节点未对应到工作流!");
        }
        JobInfo jobInfo = jobInfoMapper.selectById(jobNodeInfo.getJobId());
        if(null == jobInfo){
            throw new RRException("节点对应的工作流不存在!");
        }
        WorkMenuJobRel workMenuJobRel = workMenuJobRelMapper.selectOne(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 1).eq("work_flow_id", jobInfo.getId()));
        if(null == workMenuJobRel){
            throw new RRException("节点对应的工作流菜单不存在!");
        }

        //验证是否有权限
        iWorkMenuService.cheackProcId(jobInfo.getProcId(), ShiroUtils.getUserId(), true, jobInfo.getCreatePer(),"user",null);
        //运行状态
        if (null != jobInfo.getLastRunState() && LastRunState.RUNNING.getCode() == jobInfo.getLastRunState()) {
            throw new RRException("作业流正在运行中，请稍后重试!");
        }
        //验证引用了这个工作流的工作流依赖对应的工作流是否在运行中
        cheackWorkFlowStatus(jobInfo.getId());


        //查询其下所有的连线信息
        List<JobLink> jobLinks = jobLinkMapper.selectList(new QueryWrapper<JobLink>().eq("linke_type", 0).eq("menu_id", workMenuJobRel.getWorkMenuId()));
        //得到上游的搜的节点信息
        Set<JobLink> reUpLinks = new HashSet<>();
        Set<Integer> jobNodeIds = listUpNodes(jobLinks,jobNodeInfo.getId(),reUpLinks);
        List<JobNodeInfoDto> jobNodeInfoDtos = jobNodeInfoMapper.selectListByIds(jobNodeIds);
        Set<Integer> haveNullNodeIds = new HashSet<>();
        for (JobNodeInfoDto jobNodeInfoDto : jobNodeInfoDtos) {
            //节点是否空跑
            if(Objects.equals(1,jobNodeInfoDto.getIsNullRun())){
                haveNullNodeIds.add(jobNodeInfoDto.getId());
            }
            //路由信息
            List<JobNodeLayConfigDto> jobNodeLayConfigDtos = jobNodeLayConfigMapper.selectDtoListByJobNodeId(jobNodeInfoDto.getId(), jobNodeInfoDto.getJobId());
            jobNodeInfoDto.setJobNodeLayConfigDtoList(jobNodeLayConfigDtos);
            jobNodeInfoDto.setWorkFlowNodeName(JobType.ObjOf(jobNodeInfoDto.getTypeId()).getEnName() + "_" +  jobNodeInfoDto.getId());
            Boolean haveCodition = cheackSubNodeIsLayCondition(jobNodeInfoDto);
            if(haveCodition){
                jobNodeInfoDto.setHaveCondition(1);
            }
        }


        //获取到节点的所有的连线信息
        //生成workflow.xml
        //根据连线和路由条件，得到对应的workflow,并且保存到hdfs上
        OozieClient wc = new OozieClient(oozieConfig.getUrl());
        HdfsUtil hdfsUtil = null;
        try {
            hdfsUtil = new HdfsUtil(bdpJobConfig);
        } catch (Exception e) {
            log.error("生成hdfs帮助类异常，error:" + e.getMessage());
            throw new RRException("生成hdfs帮助类异常!");
        }
        JobUtil jobUtil = new JobUtil(hdfsUtil, wc);

        //组装数据和虚拟节点并且上传
        List<JobLink> newLinks = new ArrayList<>();
        newLinks.addAll(reUpLinks);
        List<List<JobNodeInfoDto>> allLinkedDtoList = packRouteAndVirlNode(jobNodeInfoDtos,newLinks,jobInfo,hdfsUtil);


        //根据连线和路由条件，得到对应的workflow,并且保存到hdfs上
        String oozieJobId = "";
        try {
            Properties properties = jobUtil.createOrUpdateHaveConditionJob(jobInfo.getProcId(), getQueueName(jobInfo.getProcId()), jobInfo.getId(), allLinkedDtoList, false, true, haveNullNodeIds,ReRunTypeEnum.RE_UP_NODE.getId(), null);
            oozieJobId = wc.run(properties);
            if(null != hdfsUtil){
                hdfsUtil.close();
            }
        } catch (Exception e) {
            if(null != hdfsUtil){
                hdfsUtil.close();
            }
            log.error("生成workFlowJOb异常，error:" + e.getMessage());
            throw new RRException("生成workFlowJOb异常!");
        }finally {
            if(null != hdfsUtil){
                hdfsUtil.close();
            }
        }

        //保存对应的任务的历史
        jobInfo.setLastOozieJobId(oozieJobId);
        jobInfo.setLastRunState(LastRunState.RUNNING.getCode());
        jobInfo.setLastRunType(ShiroUtils.getUserId().intValue());
        jobInfoService.updateById(jobInfo);


        //初始化job的历史记录
        List<JobNodeInfo> jobNodeInfos = jobNodeInfoMapper.selectList(new QueryWrapper<JobNodeInfo>()
                .eq("job_id", jobInfo.getId()).isNotNull("pos_x").isNotNull("pos_y"));
        String cronStartTime = null,cronEndTime = null,cronQueueName = null;
        //查询到对应的任务配置的参数时间
        JobCronDto jobCronDto = getCronTypeByJobId(jobInfo.getId());
        if(null != jobCronDto){
            if(Objects.equals(1,jobCronDto.getIsOpenTime())){
                cronStartTime = jobCronDto.getSysStartTime();
                cronEndTime = jobCronDto.getSysEndTime();
            }
            cronQueueName = jobCronDto.getQueueName();
        }
        jobNodeInfoService.initJobHistory(PojoUtils.listConvert(JobLinkDto.class, jobLinks), PojoUtils.listConvert(JobNodeInfoDto.class, jobNodeInfos), "手动运行初始化数据", jobInfo.getId(), oozieJobId,ReRunTypeEnum.RE_UP_NODE.getId(),jobNodeInfo.getId(),ShiroUtils.getUserId().intValue(), cronStartTime, cronEndTime, cronQueueName);


        //查询状态，让其进行回调，更新job任务的日志状态
        int count = 10;
        while (count > 0){
            ThreadUtil.sleep(100);
            try {
                String s = jobRestFeignApi.pollingWorkFlow(jobInfo.getId(), oozieJobId);
                if("success".equals(s)){
                    count = 0;
                }
            }catch (Exception e){
                log.error("请求运行节点上游轮询异常！error:{}",e.getMessage());
            }
            count--;
        }
    }



    @Override
    public void runDownJob(WorkMenuDto workMenuDto) {
        //运行当前节点下游节点
        if(null == workMenuDto.getId()){
            throw new RRException("节点id必须传递!");
        }
        JobNodeInfo jobNodeInfo = jobNodeInfoMapper.selectById(workMenuDto.getId());
        if(null == jobNodeInfo){
            throw new RRException("节点不存在!");
        }
        if(null == jobNodeInfo.getPosX() || null == jobNodeInfo.getPosY()){
            throw new RRException("节点未对应到工作流!");
        }
        JobInfo jobInfo = jobInfoMapper.selectById(jobNodeInfo.getJobId());
        if(null == jobInfo){
            throw new RRException("节点对应的工作流不存在!");
        }
        WorkMenuJobRel workMenuJobRel = workMenuJobRelMapper.selectOne(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 1).eq("work_flow_id", jobInfo.getId()));
        if(null == workMenuJobRel){
            throw new RRException("节点对应的工作流菜单不存在!");
        }

        //验证是否有权限
        iWorkMenuService.cheackProcId(jobInfo.getProcId(), ShiroUtils.getUserId(), true, jobInfo.getCreatePer(),"user",null);
        //运行状态
        if (null != jobInfo.getLastRunState() && LastRunState.RUNNING.getCode() == jobInfo.getLastRunState()) {
            throw new RRException("作业流正在运行中，请稍后重试!");
        }
        //验证引用了这个工作流的工作流依赖对应的工作流是否在运行中
        cheackWorkFlowStatus(jobInfo.getId());


        //查询其下所有的连线信息
        List<JobLink> jobLinks = jobLinkMapper.selectList(new QueryWrapper<JobLink>().eq("linke_type", 0).eq("menu_id", workMenuJobRel.getWorkMenuId()));
        //得到上游的搜的节点信息
        Set<JobLink> jobLinkSet = new HashSet<>();
        Set<Integer> jobNodeIds = listDownNodes(jobLinks,jobNodeInfo.getId(),jobLinkSet);
        List<JobNodeInfoDto> jobNodeInfoDtos = jobNodeInfoMapper.selectListByIds(jobNodeIds);
        Set<Integer> haveNullNodeIds = new HashSet<>();
        for (JobNodeInfoDto jobNodeInfoDto : jobNodeInfoDtos) {
            //节点是否空跑
            if(Objects.equals(1,jobNodeInfoDto.getIsNullRun())){
                haveNullNodeIds.add(jobNodeInfoDto.getId());
            }
            //路由信息
            List<JobNodeLayConfigDto> jobNodeLayConfigDtos = jobNodeLayConfigMapper.selectDtoListByJobNodeId(jobNodeInfoDto.getId(), jobNodeInfoDto.getJobId());
            jobNodeInfoDto.setJobNodeLayConfigDtoList(jobNodeLayConfigDtos);
            jobNodeInfoDto.setWorkFlowNodeName(JobType.ObjOf(jobNodeInfoDto.getTypeId()).getEnName() + "_" +  jobNodeInfoDto.getId());
            Boolean haveCodition = cheackSubNodeIsLayCondition(jobNodeInfoDto);
            if(haveCodition){
                jobNodeInfoDto.setHaveCondition(1);
            }
        }


        //获取到节点的所有的连线信息
        //生成workflow.xml
        //根据连线和路由条件，得到对应的workflow,并且保存到hdfs上
        OozieClient wc = new OozieClient(oozieConfig.getUrl());
        HdfsUtil hdfsUtil = null;
        try {
            hdfsUtil = new HdfsUtil(bdpJobConfig);
        } catch (Exception e) {
            log.error("生成hdfs帮助类异常，error:" + e.getMessage());
            throw new RRException("生成hdfs帮助类异常!");
        }
        JobUtil jobUtil = new JobUtil(hdfsUtil, wc);

        //组装数据和虚拟节点并且上传
        List<JobLink> newJobLinks = new ArrayList<>();
        newJobLinks.addAll(jobLinkSet);
        List<List<JobNodeInfoDto>> allLinkedDtoList = packRouteAndVirlNode(jobNodeInfoDtos,newJobLinks,jobInfo,hdfsUtil);


        //根据连线和路由条件，得到对应的workflow,并且保存到hdfs上
        String oozieJobId = "";
        try {
            Properties properties = jobUtil.createOrUpdateHaveConditionJob(jobInfo.getProcId(), getQueueName(jobInfo.getProcId()), jobInfo.getId(), allLinkedDtoList, false, true, haveNullNodeIds,ReRunTypeEnum.RE_DOWN_NODE.getId(), null);
            oozieJobId = wc.run(properties);
            if(null != hdfsUtil){
                hdfsUtil.close();
            }
        } catch (Exception e) {
            if(null != hdfsUtil){
                hdfsUtil.close();
            }
            log.error("生成workFlowJOb异常，error:" + e.getMessage());
            throw new RRException("生成workFlowJOb异常!");
        }finally {
            if(null != hdfsUtil){
                hdfsUtil.close();
            }
        }

        //保存对应的任务的历史
        jobInfo.setLastOozieJobId(oozieJobId);
        jobInfo.setLastRunState(LastRunState.RUNNING.getCode());
        jobInfo.setLastRunType(ShiroUtils.getUserId().intValue());
        jobInfoService.updateById(jobInfo);


        //初始化job的历史记录
        List<JobNodeInfo> jobNodeInfos = jobNodeInfoMapper.selectList(new QueryWrapper<JobNodeInfo>()
                .eq("job_id", jobInfo.getId()).isNotNull("pos_x").isNotNull("pos_y"));
        String cronStartTime = null,cronEndTime = null,cronQueueName = null;
        //查询到对应的任务配置的参数时间
        JobCronDto jobCronDto = getCronTypeByJobId(jobInfo.getId());
        if(null != jobCronDto){
            if(Objects.equals(1,jobCronDto.getIsOpenTime())){
                cronStartTime = jobCronDto.getSysStartTime();
                cronEndTime = jobCronDto.getSysEndTime();
            }
            cronQueueName = jobCronDto.getQueueName();
        }
        jobNodeInfoService.initJobHistory(PojoUtils.listConvert(JobLinkDto.class, jobLinks), PojoUtils.listConvert(JobNodeInfoDto.class, jobNodeInfos), "手动运行初始化数据", jobInfo.getId(), oozieJobId,ReRunTypeEnum.RE_DOWN_NODE.getId(),jobNodeInfo.getId(),ShiroUtils.getUserId().intValue(), cronStartTime, cronEndTime, cronQueueName);


        //查询状态，让其进行回调，更新job任务的日志状态
        int count = 10;
        while (count > 0){
            ThreadUtil.sleep(100);
            try {
                String s = jobRestFeignApi.pollingWorkFlow(jobInfo.getId(), oozieJobId);
                if("success".equals(s)){
                    count = 0;
                }
            }catch (Exception e){
                log.error("请求运行节点下游轮询异常！error:{}",e.getMessage());
            }
            count--;
        }
    }

    @Override
    public Set<Integer> listDownNodes(List<JobLink> jobLinks, Integer currentNodeId, Set<JobLink> jobLinkSet) {
        Set<Integer> restList =  new HashSet<>();
        restList.add(currentNodeId);
        if(CollectionUtil.isNotEmpty(jobLinks)){
            List<JobLink> tempList = jobLinks.stream().filter(jobLink -> Objects.equals(jobLink.getSrcNodeId(), currentNodeId)).collect(Collectors.toList());
            if(CollectionUtil.isEmpty(tempList)){
                return restList;
            }
            jobLinkSet.addAll(tempList);
            for (JobLink jobLink : tempList) {
                Set<Integer> upSets = listDownNodes(jobLinks, jobLink.getDstNodeId(), jobLinkSet);
                restList.addAll(upSets);
            }
        }
        return restList;
    }


    @Override
    public Set<Integer> listUpNodes(List<JobLink> jobLinks, Integer currentNodeId, Set<JobLink> reUpLinks) {
        Set<Integer> restList =  new HashSet<>();
        restList.add(currentNodeId);
        if(CollectionUtil.isNotEmpty(jobLinks)){
            List<JobLink> tempList = jobLinks.stream().filter(jobLink -> Objects.equals(jobLink.getDstNodeId(), currentNodeId)).collect(Collectors.toList());
            if(CollectionUtil.isEmpty(tempList)){
                return restList;
            }
            reUpLinks.addAll(tempList);
            for (JobLink jobLink : tempList) {
                Set<Integer> upSets = listUpNodes(jobLinks, jobLink.getSrcNodeId(), reUpLinks);
                restList.addAll(upSets);
            }
        }
        return restList;
    }

    @Override
    public PageUtils queryPage(DevOpsQO devOpsQO) {
        //查询到当前用户可以看到的所有的项目
        Long userId = ShiroUtils.getUserId();
        Integer tenantId = ShiroUtils.getTenantId();
        Set<Integer> lists = new HashSet<>();
        if(sysUserRoleService.superAdmin(userId)){
            if(null != devOpsQO.getProcId() && !Objects.equals(0,devOpsQO.getProcId())){
                lists.add(devOpsQO.getProcId());
            }else{
                List<ProcInfo> allProcs = procInfoMapper.listProcs(tenantId,null);
                if(CollectionUtil.isEmpty(allProcs)){
                    lists.add(0);
                }else{
                    lists.addAll(allProcs.stream().map(ProcInfo::getId).collect(Collectors.toList()));
                }
            }
        }else{
            List<ProcUser> procUsers = procUserMapper.selectList(new QueryWrapper<ProcUser>().eq("user_id", userId));
            if(CollectionUtil.isEmpty(procUsers)){
                lists.add(0);
            }else{
                Set<Integer> procIds = procUsers.stream().map(ProcUser::getProcId).collect(Collectors.toSet());
                //判断这些项目是否是当前租户的
                List<ProcInfo> allProcs = procInfoMapper.listProcs(tenantId,procIds);
                if(CollectionUtil.isEmpty(allProcs)){
                    lists.add(0);
                }else{
                    Set<Integer> allProcIds = allProcs.stream().map(ProcInfo::getId).collect(Collectors.toSet());
                    if(null != devOpsQO.getProcId() && !Objects.equals(0,devOpsQO.getProcId())){
                        if(allProcIds.contains(devOpsQO.getProcId())){
                            lists.add(devOpsQO.getProcId());
                        }else{
                            lists.add(0);
                        }
                    }else{
                        lists.addAll(allProcIds);
                    }
                }
            }
        }
        Map<String, Object> params  = new HashMap<>();
        if(CollectionUtil.isNotEmpty(lists)){
            params.put("procIds",lists);
        }
        if(StringUtils.isNotEmpty(devOpsQO.getName())){
            params.put("name",devOpsQO.getName());
        }
        if(null != devOpsQO.getLastRunState()){
            params.put("lastRunState",devOpsQO.getLastRunState());
        }
        if(null != devOpsQO.getIsOutHours() && 0 != devOpsQO.getIsOutHours()){
            params.put("isOutHours",devOpsQO.getIsOutHours());
        }
        if(null != devOpsQO.getCreatePer() && 0 != devOpsQO.getCreatePer()){
            params.put("createPer",devOpsQO.getCreatePer());
        }
        IPage<WorkMenu> infoIPage = workMenuMapper.listAndTime((Page<WorkMenu>)new QueryBean<WorkMenu, PageBean>().getPage(devOpsQO),params);
//        IPage<WorkMenu> infoIPage = workMenuMapper.selectPage(new QueryBean<WorkMenu, PageBean>().getPage(devOpsQO)
//                , new QueryWrapper<WorkMenu>()
//                        .in("work_type", 3,4)
//                        .in(CollectionUtil.isNotEmpty(lists),"proc_id", lists)
//                        .like(StringUtils.isNotEmpty(devOpsQO.getName()),"name", devOpsQO.getName())
//        .orderByDesc("id"));

        return new PageUtils(infoIPage);
    }

    @Override
    public List<Long> listUserIds(Long userId, Integer tenantId) {
        Set<Integer> lists = new HashSet<>();
        if(sysUserRoleService.superAdmin(userId)){
            List<ProcInfo> allProcs = procInfoMapper.listProcs(tenantId,null);
            if(CollectionUtil.isEmpty(allProcs)){
                lists.add(0);
            }else{
                lists.addAll(allProcs.stream().map(ProcInfo::getId).collect(Collectors.toList()));
            }
        }else{
            List<ProcUser> procUsers = procUserMapper.selectList(new QueryWrapper<ProcUser>().eq("user_id", userId));
            if(CollectionUtil.isEmpty(procUsers)){
                lists.add(0);
            }else{
                Set<Integer> procIds = procUsers.stream().map(ProcUser::getProcId).collect(Collectors.toSet());
                //判断这些项目是否是当前租户的
                List<ProcInfo> allProcs = procInfoMapper.listProcs(tenantId,procIds);
                if(CollectionUtil.isEmpty(allProcs)){
                    lists.add(0);
                }else{
                    lists.addAll(allProcs.stream().map(ProcInfo::getId).collect(Collectors.toSet()));
                }
            }
        }
        return workMenuMapper.listUserIds(lists);
    }

    @Override
    public JobCronDto getCronTypeByJobId(Integer jobId) {
        //根据id查询对应的参数
        JobCronDto jobCronDto = new JobCronDto();
        Integer scheduleOrInterval = 0;
        String cronName = null;
        String queueName = null;
        Integer isOpenTime = 0;
        String sysStartTime = null;
        String sysEndTime = null;
        Double sysDurTime = 0.0;
        JobInfo jobInfo = jobInfoMapper.selectById(jobId);
        if(null != jobInfo){
            ProcInfo procInfo = procInfoMapper.selectAndQueueNameById(jobInfo.getProcId());
            if(null != procInfo){
                queueName = procInfo.getQueueName();
            }
        }
        List<WorkMenuJobRel> workMenuJobRels = workMenuJobRelMapper.selectList(new QueryWrapper<WorkMenuJobRel>().eq("work_flow_id", jobId).eq("work_type", 1));
        if(CollectionUtil.isNotEmpty(workMenuJobRels)){
            List<JobNodeConf> jobNodeConfs = jobNodeConfMapper.selectList(new QueryWrapper<JobNodeConf>()
                    .eq("job_node_id", workMenuJobRels.get(0).getWorkMenuId())
                    .eq("job_type", 1).in("`key`", "scheduleOrInterval","schedule_type","expression"
                            ,"cronType","cronUrl"));
            if(CollectionUtil.isNotEmpty(jobNodeConfs)){
                JobNodeConf jobNodeConf3 = jobNodeConfs.stream().filter(o -> Objects.equals(o.getKey(), "cronType")).findAny().orElse(null);
                if(jobNodeConf3 != null && Objects.equals("2",jobNodeConf3.getValue())){//crontab表达式
                    JobNodeConf jobNodeConf = jobNodeConfs.stream().filter(o -> Objects.equals(o.getKey(), "schedule_type")).findAny().orElse(null);
                    if(jobNodeConf != null){
                        if(StringUtils.isNotEmpty(jobNodeConf.getValue())){
                            if(Objects.equals(jobNodeConf.getValue(),String.valueOf(ScheduleType.REPEAT.getCode()))){
                                scheduleOrInterval = -3;//默认为定时
                                cronName = "crontab周期";
                            }
                        }
                    }
                }else{
                    JobNodeConf jobNodeConf = jobNodeConfs.stream().filter(o -> Objects.equals(o.getKey(), "schedule_type")).findAny().orElse(null);
                    if(null != jobNodeConf){
                        if(StringUtils.isNotEmpty(jobNodeConf.getValue())){
                            if(Objects.equals(jobNodeConf.getValue(),String.valueOf(ScheduleType.REPEAT.getCode()))){
                                JobNodeConf jobNodeConf1 = jobNodeConfs.stream().filter(o -> Objects.equals(o.getKey(), "scheduleOrInterval")).findAny().orElse(null);
                                if(null != jobNodeConf1 && StringUtils.isNotEmpty(jobNodeConf1.getValue())){
                                    scheduleOrInterval = Integer.valueOf(jobNodeConf1.getValue());
                                    if(Objects.equals(scheduleOrInterval,ScheduleRepeatType.REPEAT.getCode())){
                                        scheduleOrInterval = -1;
                                    }
                                    if(Objects.equals(scheduleOrInterval,ScheduleRepeatType.INTERVAL.getCode())){
                                        scheduleOrInterval = -2;
                                    }
                                }
                                //对应的表达式
                                JobNodeConf jobNodeConf2 = jobNodeConfs.stream().filter(o -> Objects.equals(o.getKey(), "expression")).findAny().orElse(null);
                                if(null != jobNodeConf2){
                                    String expression = jobNodeConf2.getValue();
                                    if(StringUtils.isNotEmpty(expression)){
                                        JSONObject jsonObject = JSONObject.parseObject(expression);
                                        if(null != jsonObject){
                                            cronName = expressCrons(jsonObject.getString("type"));
                                        }
                                    }
                                }
                            }
                        }
                    }
                }
            }

            List<JobNodeConf> jobNodeConfs1 = jobNodeConfMapper.selectList(new QueryWrapper<JobNodeConf>()
                    .eq("job_node_id", workMenuJobRels.get(0).getWorkMenuId())
                    .eq("job_type", 1).in("`key`", "isOpenTime","sysStartTime","sysEndTime","sysDurTime"));
            if(CollectionUtil.isNotEmpty(jobNodeConfs1)){
                for (JobNodeConf jobNodeConf : jobNodeConfs1) {
                    if("isOpenTime".equals(jobNodeConf.getKey())){
                        if(StringUtils.isNotEmpty(jobNodeConf.getValue())){
                            isOpenTime = Integer.valueOf(jobNodeConf.getValue());
                        }
                    }
                    if("sysStartTime".equals(jobNodeConf.getKey())){
                        if(StringUtils.isNotEmpty(jobNodeConf.getValue())){
                            sysStartTime = jobNodeConf.getValue();
                        }
                    }
                    if("sysEndTime".equals(jobNodeConf.getKey())){
                        if(StringUtils.isNotEmpty(jobNodeConf.getValue())){
                            sysEndTime = jobNodeConf.getValue();
                        }
                    }
                    if("sysDurTime".equals(jobNodeConf.getKey())){
                        if(StringUtils.isNotEmpty(jobNodeConf.getValue())){
                            sysDurTime = Double.valueOf(jobNodeConf.getValue());
                        }
                    }
                }
            }
        }
        jobCronDto.setScheduleOrInterval(scheduleOrInterval);
        jobCronDto.setCronName(cronName);
        jobCronDto.setQueueName(queueName);
        jobCronDto.setIsOpenTime(isOpenTime);
        jobCronDto.setSysDurTime(sysDurTime);
        jobCronDto.setSysStartTime(sysStartTime);
        jobCronDto.setSysEndTime(sysEndTime);
        return jobCronDto;
    }

    private String expressCrons(String type) {
        String cronName = null;
        switch (type) {
            case "hour": {
                cronName = "时周期";
            }
            break;
            case "day": {
                cronName = "日周期";
            }
            break;
            case "week": {
                cronName = "周周期";
            }
            break;
            case "month": {
                cronName = "月周期";
            }
            break;
            default:
                break;
        }
        return cronName;
    }

    @Override
    public JobCronDto getCronTypeByDepId(Integer depId) {
        JobCronDto jobCronDto = new JobCronDto();
        //根据id查询
        //默认不是定时
        Integer scheduleOrInterval = 0;
        String cronName = null;
        String queueName = null;
        WorkFlowDep workFlowDep = workFlowDepService.getById(depId);
        if(null != workFlowDep){
            ProcInfo procInfo = procInfoMapper.selectAndQueueNameById(workFlowDep.getProcId());
            if(null != procInfo){
                queueName = procInfo.getQueueName();
            }
            List<JobNodeConf> jobNodeConfs = jobNodeConfMapper.selectList(new QueryWrapper<JobNodeConf>().eq("job_node_id", workFlowDep.getWorkMenuId())
                    .eq("job_type", 1).in("`key`", "scheduleOrInterval","schedule_type","expression","cronType","cronUrl"));
            if(CollectionUtil.isNotEmpty(jobNodeConfs)){
                JobNodeConf jobNodeConf = jobNodeConfs.stream().filter(o -> Objects.equals(o.getKey(), "schedule_type")).findAny().orElse(null);
                if(null != jobNodeConf){
                    if(StringUtils.isNotEmpty(jobNodeConf.getValue())){
                        if(Objects.equals(jobNodeConf.getValue(),String.valueOf(ScheduleType.REPEAT.getCode()))){
                            JobNodeConf jobNodeConf3 = jobNodeConfs.stream().filter(o -> Objects.equals(o.getKey(), "cronType")).findAny().orElse(null);
                            if(jobNodeConf3 != null && Objects.equals("2",jobNodeConf3.getValue())) {//crontab表达式
                                scheduleOrInterval = -3;//默认为定时
                                cronName = "crontab周期";
                            }else{
                                JobNodeConf jobNodeConf1 = jobNodeConfs.stream().filter(o -> Objects.equals(o.getKey(), "scheduleOrInterval")).findAny().orElse(null);
                                if(null != jobNodeConf1 && StringUtils.isNotEmpty(jobNodeConf1.getValue())){
                                    scheduleOrInterval = Integer.valueOf(jobNodeConf1.getValue());
                                    if(Objects.equals(scheduleOrInterval,ScheduleRepeatType.REPEAT.getCode())){
                                        scheduleOrInterval = -1;
                                    }
                                    if(Objects.equals(scheduleOrInterval,ScheduleRepeatType.INTERVAL.getCode())){
                                        scheduleOrInterval = -2;
                                    }
                                }
                                //对应的表达式
                                JobNodeConf jobNodeConf2 = jobNodeConfs.stream().filter(o -> Objects.equals(o.getKey(), "expression")).findAny().orElse(null);
                                if(null != jobNodeConf2){
                                    String expression = jobNodeConf2.getValue();
                                    if(StringUtils.isNotEmpty(expression)){
                                        JSONObject jsonObject = JSONObject.parseObject(expression);
                                        if(null != jsonObject){
                                            cronName = expressCrons(jsonObject.getString("type"));
                                        }
                                    }
                                }
                            }
                        }
                    }
                }
            }
        }
        jobCronDto.setScheduleOrInterval(scheduleOrInterval);
        jobCronDto.setCronName(cronName);
        jobCronDto.setQueueName(queueName);
        return jobCronDto;
    }



    private void cheackWorkFlowStatus(Integer jobId) {
        if(null != jobId){
            WorkMenuJobRel workMenuJobRel = workMenuJobRelMapper.selectOne(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 2).eq("work_flow_id", jobId));
            if(null != workMenuJobRel){
                //查询到对应的工作流依赖
                WorkFlowDep workFlowDep = workFlowDepMapper.selectOne(new QueryWrapper<WorkFlowDep>().eq("work_menu_id", workMenuJobRel.getWorkMenuId()));
                if(null != workFlowDep){
                    if(Objects.equals(LastRunState.RUNNING.getCode(),workFlowDep.getLastRunState())){
                        WorkMenu workMenu = workMenuMapper.selectById(workMenuJobRel.getWorkMenuId());
                        String workFlowLayName = workMenu == null ? "" : workMenu.getName();
                        throw new RRException("对应的作业流依赖：【"+ workFlowLayName +"】正在运行中!");
                    }
                }
            }
        }
    }

    @Override
    public void stopWorkJob(Long jobMenuId,Boolean isIngron) {
        //暂停任务
        WorkMenu workMenu = workMenuMapper.selectById(jobMenuId);
        if (null == workMenu) {
            throw new RRException("任务流菜单不存在!");
        }
        if (!Objects.equals(workMenu.getCreatePer(), ShiroUtils.getUserId().intValue()) && !isIngron) {
            throw new RRException("当前用户不能操作!");
        }

        WorkMenuJobRel workMenuJobRel = workMenuJobRelMapper.selectOne(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 1).eq("work_menu_id", jobMenuId));
        if (null == workMenuJobRel) {
            throw new RRException("作业流菜单和节点关联表数据不存在!");
        }
        JobInfo jobInfo = jobInfoMapper.selectById(workMenuJobRel.getWorkFlowId());
        if (null == jobInfo) {
            throw new RRException("任务流不存在!");
        }

        //kill掉oozie中的job
        JobUtil jobUtil = null;
        HdfsUtil hdfsUtil = null;
        if (StringUtils.isNotEmpty(jobInfo.getLastOozieJobId()) || StringUtils.isNotEmpty(jobInfo.getLastCorOozieJobId())) {
            try {
                OozieClient wc = new OozieClient(oozieConfig.getUrl());
                hdfsUtil = new HdfsUtil(bdpJobConfig);
                jobUtil = new JobUtil(hdfsUtil, wc);
                if (StringUtils.isNotEmpty(jobInfo.getLastOozieJobId())) {
                    jobUtil.stopJob(jobInfo.getLastOozieJobId());
                }
                if (StringUtils.isNotEmpty(jobInfo.getLastCorOozieJobId())) {
                    jobUtil.stopJob(jobInfo.getLastCorOozieJobId());
                }
            } catch (Exception e) {
                log.error("停止任务失败。error:" + e.getMessage());
                throw new RRException("停止任务失败!");
            }finally {
                if(null != hdfsUtil){
                    hdfsUtil.close();
                }
            }
        }

        //改变其下节点的运行状态
        //1.改变工作流的状态
        jobInfoService.updateStatus(jobInfo.getId());
        //2.暂停调度的工作流
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        executorService.execute(()->{
            //暂停调度的工作流
            jobInfoService.stopAutoJob(jobInfo.getId());
        });

        //和停止调度方式
        JobNodeConf jobNodeConf = jobNodeConfMapper.selectOne(new QueryWrapper<JobNodeConf>().eq("`key`", "schedule_type").eq("job_type", 1).eq("job_node_id", jobMenuId));
        if (null != jobNodeConf && "1".equals(jobNodeConf.getValue())) {
            jobNodeConf.setValue("0");
            jobNodeConfMapper.updateById(jobNodeConf);
        }
        //改变任务的运行状态
        if(Objects.equals(LastRunState.RUNNING.getCode(),jobInfo.getLastRunState())){
            jobInfo.setLastRunState(LastRunState.FAIL.getCode());
        }
        jobInfo.setLastCorOozieJobId(null);
        jobInfo.setLastOozieJobId(null);
        jobInfoMapper.updateById(jobInfo);
    }

    @Override
    public void stopCurrentWorkJob(Long jobMenuId) {
        //暂停任务
        WorkMenu workMenu = workMenuMapper.selectById(jobMenuId);
        if (null == workMenu) {
            throw new RRException("任务流菜单不存在!");
        }
        if (!Objects.equals(workMenu.getCreatePer(), ShiroUtils.getUserId().intValue())) {
            throw new RRException("当前用户不能操作!");
        }

        WorkMenuJobRel workMenuJobRel = workMenuJobRelMapper.selectOne(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 1).eq("work_menu_id", jobMenuId));
        if (null == workMenuJobRel) {
            throw new RRException("作业流菜单和节点关联表数据不存在!");
        }
        JobInfo jobInfo = jobInfoMapper.selectById(workMenuJobRel.getWorkFlowId());
        if (null == jobInfo) {
            throw new RRException("任务流不存在!");
        }

        //kill掉oozie中的job
        JobUtil jobUtil = null;
        HdfsUtil hdfsUtil = null;
        if (StringUtils.isNotEmpty(jobInfo.getLastOozieJobId())) {
            try {
                OozieClient wc = new OozieClient(oozieConfig.getUrl());
                hdfsUtil = new HdfsUtil(bdpJobConfig);
                jobUtil = new JobUtil(hdfsUtil, wc);
                if (StringUtils.isNotEmpty(jobInfo.getLastOozieJobId())) {
                    jobUtil.stopJob(jobInfo.getLastOozieJobId());
                }
            } catch (Exception e) {
                log.error("停止任务失败。error:" + e.getMessage());
                throw new RRException("停止任务失败!");
            }finally {
                if(null != hdfsUtil){
                    hdfsUtil.close();
                }
            }
        }

        //改变其下节点的运行状态
        //1.改变工作流的状态
        jobInfoService.updateStatus(jobInfo.getId());
        //2.暂停调度的工作流
        //暂停调度的工作流
        jobInfoService.stopAutoJob(jobInfo.getId());


        //和停止调度方式-不暂停定时调度（只停止当前运行）
//        JobNodeConf jobNodeConf = jobNodeConfMapper.selectOne(new QueryWrapper<JobNodeConf>().eq("`key`", "schedule_type").eq("job_type", 1).eq("job_node_id", jobMenuId));
//        if (null != jobNodeConf && "1".equals(jobNodeConf.getValue())) {
//            jobNodeConf.setValue("0");
//            jobNodeConfMapper.updateById(jobNodeConf);
//        }
        //改变任务的运行状态
        if(Objects.equals(LastRunState.RUNNING.getCode(),jobInfo.getLastRunState())){
            jobInfo.setLastRunState(LastRunState.STOP.getCode());
        }else{
            throw new RRException("当前作业流未运行!");
        }
        jobInfo.setLastOozieJobId(null);
        jobInfoMapper.updateById(jobInfo);

        //kill下面的任务的运行情况
        ExecutorService executorService1 = Executors.newSingleThreadExecutor();
        executorService1.execute(()->{
            killCurrnetJob(jobInfo.getLastOozieJobId(),jobInfo.getId());
        });


        //webscoket推送状态
        WorkFlowStatusVO workFlowStatusVO = new WorkFlowStatusVO();
        workFlowStatusVO.setId(Long.valueOf(jobInfo.getId()));
        workFlowStatusVO.setStatus(LastRunState.STOP.getCode());
        List<JobNodeInfo> jobNodeInfos = jobNodeInfoMapper.selectList(new LambdaQueryWrapper<JobNodeInfo>()
                .eq(JobNodeInfo::getJobId,jobInfo.getId()).isNotNull(JobNodeInfo::getPosX).isNotNull(JobNodeInfo::getPosY));
        //查询到其下都有的节点，都是未执行
        if (CollectionUtil.isNotEmpty(jobNodeInfos)) {
            List<WorkFlowStatusVO> nodeList = new ArrayList<>();
            for (JobNodeInfo jobNodeInfo : jobNodeInfos) {
                WorkFlowStatusVO workFlowStatusVO1 = new WorkFlowStatusVO();
                workFlowStatusVO1.setId(Long.valueOf(jobNodeInfo.getId()));
                workFlowStatusVO1.setStatus(LastRunState.STOP.getCode());
                nodeList.add(workFlowStatusVO1);
            }
            workFlowStatusVO.setNodeList(nodeList);
        }
        pushWebStatus(jobMenuId,workFlowStatusVO);
    }

    private void killCurrnetDepJob(String lastOozieId, Long depId) {
        //查询到这个作业流依赖的工作流日志
        List<String> oozieIds = jobDepRunHistoryMapper.selectByDepId(depId);
        if(CollectionUtil.isEmpty(oozieIds)){
            oozieIds = new ArrayList<>();
        }
        if(StringUtils.isNotEmpty(lastOozieId)){
            oozieIds.add(lastOozieId);
        }
        //查询到这个oozieId的所有的日志
        if(CollectionUtil.isEmpty(oozieIds)){
            return;
        }
        //先用depIds 换取到oozieIds
        List<String> oIds = jobRunHistoryService.selectOozieIdsByDepOozieIds(oozieIds);
        if(CollectionUtil.isEmpty(oIds)){
            return;
        }
        List<LogAndIdDto> logs = jobNodeRunHistoryService.limitListsByOOzieIds(oIds);
        if(CollectionUtil.isEmpty(logs)){
            return;
        }
        String errors = "\"status\":\"OK\"";
        List<LogAndIdDto> tmpLogs = logs.stream().distinct().filter(o -> !o.getLog().contains(errors)).collect(Collectors.toList());
        List<String> needLogs = new ArrayList<>();
        if(CollectionUtil.isNotEmpty(tmpLogs)){
            needLogs.addAll(tmpLogs.stream().sorted(Comparator.comparing(LogAndIdDto :: getId).reversed()).map(LogAndIdDto::getLog).distinct().collect(Collectors.toList()));
        }
        if(CollectionUtil.isEmpty(needLogs)){
            return;
        }
        //得到applictionIds
        if(CollectionUtil.isNotEmpty(needLogs)){
            List<String> appIds = wapperLogs(needLogs);
            if(CollectionUtil.isNotEmpty(appIds)){
                KillSparkTaskUtils.killSparkTasks(appIds,dataDevelopmentConfig.getSparkHost(),
                        dataDevelopmentConfig.getSparkPort(),
                        dataDevelopmentConfig.getSparkUserName(),
                        dataDevelopmentConfig.getSparkPwd(),depId.intValue(),2);
            }
        }
    }

    private void killCurrnetJob(String lastOozieJobId, Integer jobId) {
        List<LogAndIdDto> logs = jobNodeRunHistoryService.limitListsByJobId(jobId);
        List<String> needLogs = new ArrayList<>();
        String errors = "\"status\":\"OK\"";
        if(StringUtils.isNotEmpty(lastOozieJobId)){
            List<String> idLogs =jobNodeRunHistoryService.listByOozieId(lastOozieJobId);
            if(CollectionUtil.isNotEmpty(idLogs)){
                Set<String> tmpLogs = idLogs.stream().distinct().filter(o -> !o.contains(errors)).collect(Collectors.toSet());
                if(CollectionUtil.isNotEmpty(tmpLogs)){
                    needLogs.addAll(tmpLogs);
                }
            }
        }
        if(CollectionUtil.isNotEmpty(logs)){
            List<LogAndIdDto> tmpLogs = logs.stream().distinct().filter(o -> !o.getLog().contains(errors)).collect(Collectors.toList());
            if(CollectionUtil.isNotEmpty(tmpLogs)){
                needLogs.addAll(tmpLogs.stream().sorted(Comparator.comparing(LogAndIdDto :: getId).reversed()).map(LogAndIdDto::getLog).distinct().collect(Collectors.toList()));
            }
        }
        //得到applictionIds
        if(CollectionUtil.isNotEmpty(needLogs)){
            List<String> appIds = wapperLogs(needLogs);
            if(CollectionUtil.isNotEmpty(appIds)){
                KillSparkTaskUtils.killSparkTasks(appIds,dataDevelopmentConfig.getSparkHost(),
                        dataDevelopmentConfig.getSparkPort(),
                        dataDevelopmentConfig.getSparkUserName(),
                        dataDevelopmentConfig.getSparkPwd(),jobId,1);
            }
        }

    }

    private List<String> wapperLogs(List<String> needLogs) {
        Set<String> apps = new HashSet<>();
        if(CollectionUtil.isNotEmpty(needLogs)){
            for (String log : needLogs) {
                if(StringUtils.isNotEmpty(log)){
                    String res = wapLogs(log);
                    if(StringUtils.isNotEmpty(res)){
                        String tmps = ShellSparkUtils.shellSparkApplictionId(res);
                        if(StringUtils.isNotEmpty(tmps)){
                            apps.add(tmps);
                        }
                    }
                }
            }
        }
        return CollUtil.newArrayList(apps);
    }

    private String wapLogs(String log) {
        String appId = null;
        try {
            JSONObject logJson = JSONObject.parseObject(log);
            String consoleUrl = logJson.getString("consoleUrl");
            //"http://bigdata-test-4:8088/proxy/application_1600347239206_35663/"
            if (StringUtils.isNotEmpty(consoleUrl)) {
                String applicationId = consoleUrl.split("/")[4];
                for (String defaultFS : bdpJobConfig.getNamenodestr().split(",")) {
                    HadoopLogUtils.setDefaultFS(defaultFS);
                    String  s = HadoopLogUtils.printLog(applicationId, "yarn");
                    if (StringUtils.isNotEmpty(s)) {
                        appId = s;
                        break;
                    } else {
                        System.out.println("日志解析错误defaultFS:"+defaultFS+",applicationId:"+applicationId+",owner:yarn");
                    }
                    s = HadoopLogUtils.printLog(applicationId, "hdfs");
                    if (StringUtils.isNotEmpty(s)) {
                        appId = s;
                        break;
                    } else {
                        System.out.println("日志解析错误defaultFS:"+defaultFS+",applicationId:"+applicationId+",owner:hdfs");
                    }
                    s = HadoopLogUtils.printLog(applicationId, "spark");
                    if (StringUtils.isNotEmpty(s)) {
                        appId = s;
                        break;
                    } else {
                        System.out.println("日志解析错误defaultFS:"+defaultFS+",applicationId:"+applicationId+",owner:spark");
                    }
                }
            }
        } catch (Exception e) {
            System.out.println("无效日志！");
            e.printStackTrace();
        }
        return appId;
    }


    @Override
    public void runWorkFlowJob(Long jobWorkMenuId) {
        WorkFlowDep workFlowDep = workFlowDepMapper.selectOne(new QueryWrapper<WorkFlowDep>().eq("work_menu_id", jobWorkMenuId));
        if (null == workFlowDep) {
            throw new RRException("工作流依赖不存在：jobWorkMenuId" + jobWorkMenuId + "!");
        }
        if (null != workFlowDep.getLastRunState() && LastRunState.RUNNING.getCode() == workFlowDep.getLastRunState()) {
            throw new RRException("工作流依赖正在运行中，请稍后重试!");
        }
        //如果是禁用状态，请先启用工作流依赖
        if (1 != workFlowDep.getJobLayStatus()) {
            throw new RRException("请先启用工作流依赖!");
        }
        //验证是否有权限
        iWorkMenuService.cheackProcId(workFlowDep.getProcId(), ShiroUtils.getUserId(), true, workFlowDep.getCreatePer(),"user",null);

        //查询对应的调度方式和系统参数
        List<JobNodeConf> jobNodeConfs = jobNodeConfMapper.selectList(new QueryWrapper<JobNodeConf>()
                .eq("job_node_id", jobWorkMenuId).eq("job_type", 1));
        JobNodeConfig jobNodeConfig = null;
        if (CollectionUtil.isNotEmpty(jobNodeConfs)) {
            JobNodeConf olderJobNodeConf = null;
            List<JobNodeConf> newJobNodeConfs = new ArrayList<>();
            for (JobNodeConf jobNodeConf : jobNodeConfs) {
                if ("sysParam".equals(jobNodeConf.getKey())) {
                    olderJobNodeConf = jobNodeConf;
                } else {
                    newJobNodeConfs.add(jobNodeConf);
                }
            }
            if (CollectionUtil.isNotEmpty(newJobNodeConfs)) {
                jobNodeConfig = JobNodeConfig.listToJobNodeConfig(newJobNodeConfs);
            }
            //系统参数
            if (null != olderJobNodeConf) {
                if (null == jobNodeConfig) {
                    jobNodeConfig = new JobNodeConfig();
                }
                JSONArray jsonArray = JSONArray.parseArray(olderJobNodeConf.getValue());
                if (CollectionUtil.isNotEmpty(jsonArray)) {
                    List<Map<String, Object>> restList = new ArrayList<>();
                    for (Object obj : jsonArray) {
                        Map<String, Object> values = JSONObject.parseObject(obj.toString()).getInnerMap();
                        restList.add(values);
                    }
                    jobNodeConfig.setSysParam(restList);
                }
            }
        }
        //运行对应的任务
        //查询其下所有的连线信息
        List<JobLink> jobLinks = jobLinkMapper.selectList(new QueryWrapper<JobLink>().eq("linke_type", 1).eq("menu_id", jobWorkMenuId));
        //获取其对应的job任务信息
        List<WorkMenuJobRel> workMenuJobRels = workMenuJobRelMapper.selectList(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 2).eq("work_menu_id", jobWorkMenuId));
        if (CollectionUtil.isEmpty(workMenuJobRels)) {
            return;
        }
        Set<Integer> ids = new HashSet<>();
        for (WorkMenuJobRel workMenuJobRel : workMenuJobRels) {
            ids.add(workMenuJobRel.getWorkFlowId().intValue());
        }
        //得到所有的jobInfo信息
        List<JobInfo> jobInfos = jobInfoMapper.selectBatchIds(ids);
        if (CollectionUtil.isEmpty(jobInfos)) {
            return;
        }
        //获取到节点的所有的连线信息
        List<List<JobInfoDto>> allLinkedDtoList = listAllJobList(PojoUtils.listConvert(JobLinkDto.class, jobLinks), ids);

        //根据连线和路由条件，得到对应的workflow,并且保存到hdfs上
        String oozieJobId = "";
        HdfsUtil hdfsUtil = null;
        try {
            OozieClient wc = new OozieClient(oozieConfig.getUrl());
            hdfsUtil = new HdfsUtil(bdpJobConfig);
            JobUtil jobUtil = new JobUtil(hdfsUtil, wc);
            Properties conf = jobUtil.createOrUpdateJobLay(workFlowDep.getProcId(),getQueueName(workFlowDep.getProcId()), workFlowDep.getId().intValue(), allLinkedDtoList, false, false,jobNodeConfig);
            oozieJobId = wc.run(conf);
        } catch (Exception e) {
            log.error("任务流依赖生成失败!error:" + e.getMessage());
            throw new RRException("任务流依赖生成失败!");
        }finally {
            if(null != hdfsUtil){
                hdfsUtil.close();
            }
        }

        //保存对应的任务的历史
        workFlowDep.setLastOozieId(oozieJobId);
        workFlowDep.setLastRunState(LastRunState.RUNNING.getCode());
        workFlowDep.setLastRunType(ShiroUtils.getUserId().intValue());
        workFlowDepMapper.updateById(workFlowDep);


        //查询到这个工程对应的调度队列
        ProcInfo procInfo = procInfoMapper.selectAndQueueNameById(workFlowDep.getProcId());
        String queueName = procInfo == null ? "" : procInfo.getQueueName();
        //初始化工作流依赖的历史记录
        workFlowDepService.initWorkFlowDepHistory(PojoUtils.listConvert(JobLinkDto.class, jobLinks), PojoUtils.listConvert(JobInfoDto.class, jobInfos), "手动运行初始化数据", workFlowDep.getId().intValue(), oozieJobId,ShiroUtils.getUserId().intValue(),queueName);


        int count = 10;
        while (count > 0){
            ThreadUtil.sleep(50);
            try {
                String s = jobRestFeignApi.pollingWorkFlowLay(workFlowDep.getId(), oozieJobId);
                if("success".equals(s)){
                    count = 0;
                }
            }catch (Exception e){
                log.error("请求工作流依赖轮询异常！error:{}",e.getMessage());
            }
            count--;
        }

        //webscoket推送状态
        WorkFlowStatusVO workFlowStatusVO = new WorkFlowStatusVO();
        workFlowStatusVO.setId(workFlowDep.getId());
        workFlowStatusVO.setStatus(LastRunState.RUNNING.getCode());
        workFlowStatusVO.setOozieId(oozieJobId);
        //查询到其下都有的节点，都是未执行
        if (CollectionUtil.isNotEmpty(jobInfos)) {
            List<WorkFlowStatusVO> nodeList = new ArrayList<>();
            for (JobInfo jobInfo : jobInfos) {
                WorkFlowStatusVO workFlowStatusVO1 = new WorkFlowStatusVO();
                workFlowStatusVO1.setId(Long.valueOf(jobInfo.getId()));
                workFlowStatusVO1.setStatus(LastRunState.NOTRUN.getCode());
                workFlowStatusVO1.setOozieId(oozieJobId);
                nodeList.add(workFlowStatusVO1);
            }
            workFlowStatusVO.setNodeList(nodeList);
        }
        //遍历注册到这个集合中的所有的需要查询的数据
        ConcurrentHashMap<String, Session> sessionMap = WorkFlowLayStatusSocketServer.listSessionMap();
        Iterator<Map.Entry<String, Session>> entries = sessionMap.entrySet().iterator();
        while (entries.hasNext()) {
            Map.Entry<String, Session> entry = entries.next();
            String[] split = entry.getKey().split("_");
            if(null != split && split.length >= 3){
                if(split[0].equals(jobWorkMenuId.toString())){
                    if (null != workFlowStatusVO) {
                        WorkFlowLayStatusSocketServer.sendMessage(entry.getValue(), JSONObject.toJSONString(workFlowStatusVO));
                    }
                }
            }
        }

    }

    @Override
    public void closeWorkFlowJob(Long jobWorkMenuId) {
        WorkFlowDep workFlowDep = workFlowDepMapper.selectOne(new QueryWrapper<WorkFlowDep>().eq("work_menu_id", jobWorkMenuId));
        if (null == workFlowDep) {
            throw new RRException("工作流依赖不存在!");
        }
        if (workFlowDep.getJobLayStatus() == 0) {
            throw new RRException("工作流依赖已是禁用状态!");
        }


        //停止自动调度的job任务
        HdfsUtil hdfsUtil = null;
        try {
            OozieClient wc = new OozieClient(oozieConfig.getUrl());
            hdfsUtil = new HdfsUtil(bdpJobConfig);
            JobUtil jobUtil = new JobUtil(hdfsUtil, wc);
            if (StringUtils.isNotEmpty(workFlowDep.getLastCorOozieId())) {
                jobUtil.stopJob(workFlowDep.getLastCorOozieId());
                cheackJobStatusAndOozieId(wc,workFlowDep.getLastCorOozieId());
            }
            if (StringUtils.isNotEmpty(workFlowDep.getLastOozieId())) {
                jobUtil.stopJob(workFlowDep.getLastOozieId());
                cheackJobStatusAndOozieId(wc,workFlowDep.getLastOozieId());
            }
            //停止最近的10条运行中的任务
            //暂停其下所有的工作流依赖，工作流，节点
            stopAllJobDepAndJobAndJobNode(jobUtil,workFlowDep.getId(),workFlowDep.getLastCorOozieId(),workFlowDep.getLastOozieId());
        } catch (Exception e) {
            log.error("禁用工作流依赖异常!，error:" + e.getMessage());
            throw new RRException("禁用工作流依赖异常,"+e.getMessage());
        }finally {
            if(null != hdfsUtil){
                hdfsUtil.close();
            }
        }

        //kill下面的任务的运行情况
        ExecutorService executorService1 = Executors.newSingleThreadExecutor();
        executorService1.execute(()->{
            killCurrnetDepJob(workFlowDep.getLastOozieId(),workFlowDep.getId());
        });


        if(Objects.equals(LastRunState.RUNNING.getCode(),workFlowDep.getLastRunState())){
            workFlowDep.setLastRunState(LastRunState.STOP.getCode());
        }
        workFlowDep.setJobLayStatus(0);
        workFlowDep.setLastOozieId(null);
        workFlowDep.setLastCorOozieId(null);
        workFlowDepMapper.updateById(workFlowDep);
        workFlowDepMapper.updateLastCorOozied(workFlowDep.getId());
        workFlowDepMapper.updateLastOozied(workFlowDep.getId());
//        //还得改为暂停调度
//        JobNodeConf jobNodeConf = jobNodeConfMapper.selectOne(new QueryWrapper<JobNodeConf>()
//                .eq("`key`", "schedule_type")
//                .eq("job_node_id", jobWorkMenuId).eq("job_type", 1));
//        if(null != jobNodeConf){
//            jobNodeConf.setValue(String.valueOf(ScheduleType.ONECE.getCode()));
//            jobNodeConfMapper.updateById(jobNodeConf);
//        }

    }

    private void stopAllJobDepAndJobAndJobNode(JobUtil jobUtil, Long jobDepId, String lastCorOozieId, String lastOozieId) {
        if(StringUtils.isNotEmpty(lastOozieId)){
            //手动作业流依赖
            List<JobDepHistory> jobDepHistories = jobDepRunHistoryMapper.selectList(new QueryWrapper<JobDepHistory>().eq("oozie_job_id", lastOozieId));
            if(CollectionUtil.isNotEmpty(jobDepHistories)){
                for (JobDepHistory jobDepHistory : jobDepHistories) {
                    try {
                        //改变状态
                        if(Objects.equals(LastRunState.RUNNING.getCode(),jobDepHistory.getState())){
                            jobDepHistory.setState(LastRunState.STOP.getCode());
                            jobDepRunHistoryMapper.updateById(jobDepHistory);
                        }
                    } catch (Exception e){
                        log.error("停止作业流依赖异常，error:{}",e.getMessage());
                    }
                }
            }
            //改变其工作流和其节点的状态和停止
            List<JobRunHistory> jobsList = jobRunHistoryService.list(new QueryWrapper<JobRunHistory>()
                    .eq("dep_oozie_job_id", lastOozieId));
            if(CollectionUtil.isNotEmpty(jobsList)){
                for (JobRunHistory jobRunHistory : jobsList) {
                    try {
                        //改变状态
                        if(Objects.equals(LastRunState.RUNNING.getCode(),jobRunHistory.getState())){
                            jobRunHistory.setState(LastRunState.STOP.getCode());
                            jobRunHistoryService.updateById(jobRunHistory);
                        }
                        //改变其下节点的运行状态
                        List<JobNodeRunHistory> jobNodeRunHistories = jobNodeRunHistoryService.list(new QueryWrapper<JobNodeRunHistory>().eq("oozie_job_id", jobRunHistory.getOozieJobId()));
                        if(CollectionUtil.isNotEmpty(jobNodeRunHistories)){
                            for (JobNodeRunHistory jobNodeRunHistory : jobNodeRunHistories) {
                                try {
                                    if(Objects.equals(LastRunState.RUNNING.getCode(),jobNodeRunHistory.getState())){
                                        jobNodeRunHistory.setState(LastRunState.STOP.getCode());
                                        jobNodeRunHistoryService.updateById(jobNodeRunHistory);
                                    }
                                }catch (Exception e){
                                    log.error("停止作业流依赖中的任务中的节点异常，error:{}",e.getMessage());
                                }
                            }
                        }
                        //停止任务
                        jobUtil.stopJob(jobRunHistory.getOozieJobId());
                    }catch (Exception e){
                        log.error("停止作业流依赖中的任务异常，error:{}",e.getMessage());
                    }
                }
            }
        }

        //自动触发
        //查询到最近的20条历史数据
        List<JobDepHistory> jobDepHistories = jobDepRunHistoryMapper.selectByLimit20(jobDepId,5);
        if(CollectionUtil.isNotEmpty(jobDepHistories)){
            for (JobDepHistory jobDepHistory : jobDepHistories) {
                try {
                    //改变状态
                    if(Objects.equals(LastRunState.RUNNING.getCode(),jobDepHistory.getState())){
                        jobDepHistory.setState(LastRunState.STOP.getCode());
                        jobDepRunHistoryMapper.updateById(jobDepHistory);
                    }

                    //改变其工作流和其节点的状态和停止
                    List<JobRunHistory> jobsList = jobRunHistoryService.list(new QueryWrapper<JobRunHistory>()
                            .eq("dep_oozie_job_id", jobDepHistory.getOozieJobId()));
                    if(CollectionUtil.isNotEmpty(jobsList)){
                        for (JobRunHistory jobRunHistory : jobsList) {
                            try {
                                //改变状态
                                if(Objects.equals(LastRunState.RUNNING.getCode(),jobRunHistory.getState())){
                                    jobRunHistory.setState(LastRunState.STOP.getCode());
                                    jobRunHistoryService.updateById(jobRunHistory);
                                }
                                //改变其下节点的运行状态
                                List<JobNodeRunHistory> jobNodeRunHistories = jobNodeRunHistoryService.list(new QueryWrapper<JobNodeRunHistory>().eq("oozie_job_id", jobRunHistory.getOozieJobId()));
                                if(CollectionUtil.isNotEmpty(jobNodeRunHistories)){
                                    for (JobNodeRunHistory jobNodeRunHistory : jobNodeRunHistories) {
                                        try {
                                            if(Objects.equals(LastRunState.RUNNING.getCode(),jobNodeRunHistory.getState())){
                                                jobNodeRunHistory.setState(LastRunState.STOP.getCode());
                                                jobNodeRunHistoryService.updateById(jobNodeRunHistory);
                                            }
                                        }catch (Exception e){
                                            log.error("停止定时作业流依赖中的任务中的节点异常，error:{}",e.getMessage());
                                        }
                                    }
                                }
                                //停止任务
                                jobUtil.stopJob(jobRunHistory.getOozieJobId());
                            }catch (Exception e){
                                log.error("停止定时作业流依赖中的任务异常，error:{}",e.getMessage());
                            }
                        }
                    }

                    jobUtil.stopJob(jobDepHistory.getOozieJobId());

                }catch (Exception e){
                    log.error("停止定时作业流依赖中的任务异常，error:{}",e.getMessage());
                }
            }
        }

    }

    @Override
    public void openWorkFlowJob(Long jobWorkMenuId) {
        WorkFlowDep workFlowDep = workFlowDepMapper.selectOne(new QueryWrapper<WorkFlowDep>().eq("work_menu_id", jobWorkMenuId));
        if (null == workFlowDep) {
            throw new RRException("工作流依赖不存在!");
        }
        if (workFlowDep.getJobLayStatus() == 1) {
            throw new RRException("工作流依赖已是启用状态!");
        }
        //验证是否有配置对应的节点连线
        List<JobLink> jobLinks = jobLinkMapper.selectList(new QueryWrapper<JobLink>().eq("menu_id", jobWorkMenuId).eq("linke_type", 1));
        //获取其对应的job任务信息
        List<WorkMenuJobRel> workMenuJobRels = workMenuJobRelMapper.selectList(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 2).eq("work_menu_id", workFlowDep.getWorkMenuId()));
        if (CollectionUtil.isEmpty(workMenuJobRels)) {
            throw new RRException("工作流依赖未配置工作流!");
        }
        Set<Integer> ids = new HashSet<>();
        for (WorkMenuJobRel workMenuJobRel : workMenuJobRels) {
            ids.add(workMenuJobRel.getWorkFlowId().intValue());
        }
        //得到所有的jobInfo信息
        List<JobInfo> jobInfos = jobInfoMapper.selectBatchIds(ids);
        if (CollectionUtil.isEmpty(jobInfos)) {
            throw new RRException("工作流依赖配置工作流不存在!");
        }
        //如果连线下面的job有配置正常调度全部改为暂停调度
        stopAllCroJob(jobWorkMenuId);


        //查询对应的调度方式和系统参数
        List<JobNodeConf> jobNodeConfs = jobNodeConfMapper.selectList(new QueryWrapper<JobNodeConf>()
                .eq("job_node_id", jobWorkMenuId).eq("job_type", 1));
        JobNodeConfig jobNodeConfig = null;
        if (CollectionUtil.isNotEmpty(jobNodeConfs)) {
            JobNodeConf olderJobNodeConf = null;
            List<JobNodeConf> newJobNodeConfs = new ArrayList<>();
            for (JobNodeConf jobNodeConf : jobNodeConfs) {
                if ("sysParam".equals(jobNodeConf.getKey())) {
                    olderJobNodeConf = jobNodeConf;
                } else {
                    newJobNodeConfs.add(jobNodeConf);
                }
            }
            if (CollectionUtil.isNotEmpty(newJobNodeConfs)) {
                jobNodeConfig = JobNodeConfig.listToJobNodeConfig(newJobNodeConfs);
            }
            //系统参数
            if (null != olderJobNodeConf) {
                if (null == jobNodeConfig) {
                    jobNodeConfig = new JobNodeConfig();
                }
                JSONArray jsonArray = JSONArray.parseArray(olderJobNodeConf.getValue());
                if (CollectionUtil.isNotEmpty(jsonArray)) {
                    List<Map<String, Object>> restList = new ArrayList<>();
                    for (Object obj : jsonArray) {
                        Map<String, Object> values = JSONObject.parseObject(obj.toString()).getInnerMap();
                        restList.add(values);
                    }
                    jobNodeConfig.setSysParam(restList);
                }
            }
        }

        //重新获取去掉开始节点
        List<JobLinkDto> insertDataList = new ArrayList<>();
        jobLinks.forEach(jobLinkDto -> {
            if (!Objects.equals(0, jobLinkDto.getSrcNodeId())) {
                JobLinkDto jobLinkDto1 = new JobLinkDto();
                BeanUtils.copyProperties(jobLinkDto, jobLinkDto1);
                jobLinkDto1.setCreatePer(ShiroUtils.getUserId().intValue());
                jobLinkDto1.setCreateTime(DateTime.now());
                jobLinkDto1.setLinkeType(1);
                jobLinkDto1.setMenuId(jobWorkMenuId);
                insertDataList.add(jobLinkDto1);
            }
        });

        //生成对应的虚拟化开始节点和结束节点的shell
        saveStartJobNodeAndEndJobNode(workFlowDep.getProcId().longValue(), workFlowDep.getId().intValue(), WorkFlowType.WORKFLOW_LAY.getCode());

        //生成workflow的xml文件
        try {
            generatedAutoWorkFlowLayXml(insertDataList, ids, jobNodeConfig, workFlowDep.getId(), workFlowDep.getProcId());
            WorkFlowDep workFlowDepNew = workFlowDepMapper.selectById(workFlowDep.getId());
            workFlowDepNew.setJobLayStatus(1);
            workFlowDepMapper.updateById(workFlowDepNew);
        } catch (Exception e) {
            log.error("生成对应的定时调度的工作流依赖异常！error:" + e.getMessage());
            throw new RRException("生成对应的定时调度的工作流依赖异常！error:" + e.getMessage());
        }

    }

    private void stopAllCroJob(Long jobWorkMenuId) {
        List<WorkMenuJobRel> workMenuJobRels = workMenuJobRelMapper.selectList(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 2).eq("work_menu_id", jobWorkMenuId));
        if (CollectionUtil.isNotEmpty(workMenuJobRels)) {
            for (WorkMenuJobRel workMenuJobRel : workMenuJobRels) {
                //根据工作流id查询到对应的菜单
                WorkMenuJobRel workMenuJobRel1 = workMenuJobRelMapper.selectOne(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 1).eq("work_flow_id", workMenuJobRel.getWorkFlowId()));
                if(null != workMenuJobRel1){
                    //将对应的调度方式改为暂停调度
                    JobNodeConf jobNodeConf = jobNodeConfMapper.selectOne(new QueryWrapper<JobNodeConf>().eq("`key`", "schedule_type")
                            .eq("`value`", 1)
                            .eq("job_node_id", workMenuJobRel1.getWorkMenuId())
                            .eq("job_type", 1));
                    if (null != jobNodeConf) {
                        jobNodeConf.setValue("0");
                        jobNodeConfMapper.updateById(jobNodeConf);
                    }
                }

                //停止正常的调度
                JobInfo jobInfo = jobInfoMapper.selectById(workMenuJobRel.getWorkFlowId());
                if (null != jobInfo) {
                    if (StringUtils.isNotEmpty(jobInfo.getLastCorOozieJobId())) {
                        JobUtil jobUtil = null;
                        HdfsUtil hdfsUtil = null;
                        try {
                            OozieClient wc = new OozieClient(oozieConfig.getUrl());
                            hdfsUtil = new HdfsUtil(bdpJobConfig);
                            jobUtil = new JobUtil(hdfsUtil, wc);
                            jobUtil.stopJob(jobInfo.getLastCorOozieJobId());
                        } catch (Exception e) {
                            log.error("停止工作流异常!，error:" + e.getMessage());
                            throw new RRException("停止工作流异常!");
                        }finally {
                            if(null != hdfsUtil){
                                hdfsUtil.close();
                            }
                        }
                    }
                    //暂停调度
                    jobInfoService.stopAutoJob(jobInfo.getId());
                }
            }
        }
    }

    /***
     * 慎重修改，如果修改，则还得重新发布job-center服务
     * @param jobId 任务id
     * @return
     */
    @Override
    public String syncRunWorkJob(Long jobId,String depOOzieId) {
        //使用场景：1.作业流依赖，通过shell调用这个接口，启动对应的job,并且直到执行完成，才返回对应的结果给下一个节点任务
        JobInfo jobInfo = jobInfoMapper.selectById(jobId);
        if (null == jobInfo) {
            throw new RRException("job任务不存在，其id为：" + jobId + "!");
        }
        if(StringUtils.isEmpty(depOOzieId)){
            throw new RRException("作业流依赖的的oozie不存在！depOOzieId：" + depOOzieId + "!");
        }
        WorkMenuJobRel workMenuJobRel = workMenuJobRelMapper.selectOne(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 1).eq("work_flow_id", jobId));
        if (null == workMenuJobRel) {
            throw new RRException("作业流菜单和节点关联表数据不存在!");
        }
        //查询到任务的依赖关系和对应的节点关系
        //查询其下所有的连线信息
        List<JobLink> jobLinks = jobLinkMapper.selectList(new QueryWrapper<JobLink>().eq("linke_type", 0).eq("menu_id", workMenuJobRel.getWorkMenuId()));
        //查询其下所有的节点
        List<JobNodeInfo> jobNodeInfos1 = jobNodeInfoMapper.selectList(new QueryWrapper<JobNodeInfo>().eq("job_id", jobInfo.getId()).isNotNull("pos_x").isNotNull("pos_y"));
        if (CollectionUtil.isEmpty(jobNodeInfos1)) {
            throw new RRException("作业流未配置节点!");
        }
        Set<Integer> nodeIds = new HashSet<>();
        for (JobNodeInfo jobNodeInfo : jobNodeInfos1) {
            nodeIds.add(jobNodeInfo.getId());
        }
        List<JobNodeInfoDto> jobNodeInfoDtos = jobNodeInfoMapper.selectListByIds(nodeIds);
        if(CollectionUtil.isEmpty(jobNodeInfoDtos)){
            throw new RRException("工作流未配置节点!");
        }

        JobRunHistory jobRunHistory = new JobRunHistory();
        //查询当前的整个工作流的状态（并且来修改状态）
        JobDepHistory jobDepHistory = jobDepRunHistoryMapper.selectOne(new QueryWrapper<JobDepHistory>()
                .eq("oozie_job_id", depOOzieId)
        .last(" limit 1"));
        //查询到对应的历史记录
        if(null != jobDepHistory){
            WorkFlowDep workFlowDep = workFlowDepMapper.selectById(jobDepHistory.getJobDepId());
            if(null != workFlowDep){
                //查询到对应的任务配置的参数时间
                JobCronDto jobCronDto = getCronTypeByJobId(jobId.intValue());
                if(null != jobCronDto){
                    //判断是手动的还是自动的
                    if(Objects.equals(1,jobCronDto.getIsOpenTime())){
                        jobRunHistory.setCronStartTime(jobCronDto.getSysStartTime());
                        jobRunHistory.setCronEndTime(jobCronDto.getSysEndTime());
                        //如果是手动的，则时间不需要更新。如果是定时的则需要修改
                        if(!Objects.equals(workFlowDep.getLastOozieId(),depOOzieId)){
                            jobCronDto.setWorkMenuId(workMenuJobRel.getWorkMenuId());
                            //编辑下一个时间
                            jobNodeConfService.insertJobNodeConfig(jobCronDto);
                        }
                    }
                }
                //如果不是运行中（直接不提交任务）。现在这个状态有问题
//                if(!Objects.equals(LastRunState.RUNNING.getCode(),workFlowDep.getLastRunState())){
//                    return "-1";
//                }
            }
        }

        Set<Integer> haveNullNodeIds = new HashSet<>();
        for (JobNodeInfoDto jobNodeInfoDto : jobNodeInfoDtos) {
            //节点是否空跑
            if(Objects.equals(1,jobNodeInfoDto.getIsNullRun())){
                haveNullNodeIds.add(jobNodeInfoDto.getId());
            }
            //路由信息
            List<JobNodeLayConfigDto> jobNodeLayConfigDtos = jobNodeLayConfigMapper.selectDtoListByJobNodeId(jobNodeInfoDto.getId(), jobNodeInfoDto.getJobId());
            jobNodeInfoDto.setJobNodeLayConfigDtoList(jobNodeLayConfigDtos);
            jobNodeInfoDto.setWorkFlowNodeName(JobType.ObjOf(jobNodeInfoDto.getTypeId()).getEnName() + "_" +  jobNodeInfoDto.getId());
            Boolean haveCodition = cheackSubNodeIsLayCondition(jobNodeInfoDto);
            if(haveCodition){
                jobNodeInfoDto.setHaveCondition(1);
            }
        }
        //获取到节点的所有的连线信息
        //生成workflow.xml
        //根据连线和路由条件，得到对应的workflow,并且保存到hdfs上
        OozieClient wc = new OozieClient(oozieConfig.getUrl());
        HdfsUtil hdfsUtil = null;
        try {
            hdfsUtil = new HdfsUtil(bdpJobConfig);
        } catch (Exception e) {
            log.error("生成hdfs帮助类异常，error:" + e.getMessage());
            throw new RRException("生成hdfs帮助类异常!");
        }
        JobUtil jobUtil = new JobUtil(hdfsUtil, wc);

        //组装数据和虚拟节点并且上传
        List<List<JobNodeInfoDto>> allLinkedDtoList = packRouteAndVirlNode(jobNodeInfoDtos,jobLinks,jobInfo,hdfsUtil);


        //根据连线和路由条件，得到对应的workflow,并且保存到hdfs上
        String oozieJobId = "";
        try {
            Properties properties = jobUtil.createOrUpdateHaveConditionJob(jobInfo.getProcId(), getQueueName(jobInfo.getProcId()), jobInfo.getId(), allLinkedDtoList, false, false, haveNullNodeIds, ReRunTypeEnum.RE_ALL_NODE.getId(), null);
            oozieJobId = wc.run(properties);
        } catch (Exception e) {
            log.error("回调启动工作流异常，error:" + e.getMessage());
            throw new RRException("回调启动工作流异常!");
        }finally {
            if(null != hdfsUtil){
                hdfsUtil.close();
            }
        }

        //初始化对应的job的历史数据   判断是否是定时执行的（手动的oozieId如果和depOOzieId 不相等的话，就是定时的）
        jobRunHistory.setJobInfoId(jobInfo.getId());
        jobRunHistory.setNum(null);
        jobRunHistory.setParams("依赖初始化进来的工作流");
        JobLinkAndJobNodeDto jobLinkAndJobNodeDto = new JobLinkAndJobNodeDto();
        //job对应的菜单
        jobLinkAndJobNodeDto.setEdges(PojoUtils.listConvert(JobLinkDto.class,jobLinks));
        jobLinkAndJobNodeDto.setNodes(jobNodeInfoDtos);
        jobRunHistory.setNodeParams(JSONObject.toJSONString(jobLinkAndJobNodeDto));
        jobRunHistory.setOozieJobId(oozieJobId);
        jobRunHistory.setDepOozieJobId(depOOzieId);
        jobRunHistory.setOozieStatus("running");
        jobRunHistory.setState(LastRunState.RUNNING.getCode());
        ProcInfo procInfo = procInfoMapper.selectAndQueueNameById(jobInfo.getProcId());
        if(null != procInfo){
            jobRunHistory.setQueueName(procInfo.getQueueName());
        }

        jobRunHistoryService.save(jobRunHistory);

        return oozieJobId;
    }

    @Override
    public WorkFlowDepDto selectWorkFlowDep(WorkMenuDto workMenuDto) {
        WorkFlowDep workFlowDep = workFlowDepMapper.selectOne(new QueryWrapper<WorkFlowDep>().eq("work_menu_id", workMenuDto.getId()));
        if (null != workFlowDep) {
            WorkFlowDepDto workFlowDepDto = new WorkFlowDepDto();
            BeanUtils.copyProperties(workFlowDep, workFlowDepDto);
            SysUserEntity sysUserEntity = sysUserMapper.selectById(workFlowDep.getCreatePer());
            workFlowDepDto.setCreatePerName(null == sysUserEntity ? "" : sysUserEntity.getUsername());
            return workFlowDepDto;
        }
        return null;
    }

    @Override
    public JobInfoDto getJobInfo(WorkMenuDto workMenuDto) {
        if (null == workMenuDto || workMenuDto.getId() == null) {
            throw new RRException(DataDevelopmentBizExceptionEnum.PARAM_ERROR.getMessage());
        }
        WorkMenu workMenu = workMenuMapper.selectById(workMenuDto.getId());
        if (null == workMenu) {
            throw new RRException("默认作业流不存在!");
        }
        if (WorkFlowMenuType.WORKFLOW_FIELD.getCode() != workMenu.getWorkType()) {
            throw new RRException("上级目录不是默认作业流类型!");
        }
        WorkMenuJobRel workMenuJobRel = workMenuJobRelMapper.selectOne(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 1).eq("work_menu_id", workMenuDto.getId()));
        if (null == workMenuJobRel) {
            throw new RRException("默认作业流对应的任务不存在!");
        }
        JobInfo jobInfo = jobInfoMapper.selectById(workMenuJobRel.getWorkFlowId());
        if (null != jobInfo) {
            JobInfoDto jobInfoDto = new JobInfoDto();
            BeanUtils.copyProperties(jobInfo, jobInfoDto);
            SysUserEntity sysUserEntity = sysUserMapper.selectById(jobInfo.getCreatePer());
            jobInfoDto.setCreatePerName(null == sysUserEntity ? "" : sysUserEntity.getUsername());
            return jobInfoDto;
        }
        return null;
    }


    private String pingRestApiStartShell(String oozieJobId, String jobName,Integer workFlowType) {
        if(Objects.equals(workFlowType,WorkFlowType.WORKFLOW.getCode())){
            return "#!/bin/sh\n" +
                    "source /etc/profile\n" +
                    "param=${time_hour}\n" +
                    "stat_date=`date +%Y-%m-%d`\n" +
                    "if [ -z \"$param\" ]; then\n" +
                    "param=`date '+%Y-%m-%d %H:%M:%S'`\n" +
                    "fi\n" +
                    "currentOOzieId=$1 \n"+
                    "echo oozieId: ${currentOOzieId}  \n"+
                    "function run(){  \necho 'end run'" +
                    "\n}\n" +
                    "run && (curl -H \"Content-Type:application/json;charset=UTF-8\" -d \"{\"\'\"jobName\"\'\":\\\"" + jobName + "\\\"," +
                    "\"\'\"currentOOzieId\"\'\":\\\"${currentOOzieId}\\\" ," +
                    "\"\'\"oozieJobId\"\'\":\\\"" + oozieJobId + "\\\"," +
                    "\"\'\"stat_date\"\'\":\\\"${param}\\\" ," +
                    "\"\'\"isCorJob\"\'\":\\\"" + true + "\\\"}\" \"" + oozieConfig.getRest_url() + "/job_begin\") || exit 10";

        }else{
            return "#!/bin/sh\n" +
                    "source /etc/profile\n" +
                    "param=${time_hour}\n" +
                    "stat_date=`date +%Y-%m-%d`\n" +
                    "if [ -z \"$param\" ]; then\n" +
                    "param=`date '+%Y-%m-%d %H:%M:%S'`\n" +
                    "fi\n" +
                    "currentOOzieId=$1 \n"+
                    "echo oozieId: ${currentOOzieId}  \n"+
                    "function run(){  \necho 'end run'" +
                    "\n}\n" +
                    "run\ndepOOzieId=`curl -H \"Content-Type:application/json;charset=UTF-8\" -d \"{\"\'\"jobName\"\'\":\\\"" + jobName + "\\\"," +
                    "\"\'\"currentOOzieId\"\'\":\\\"${currentOOzieId}\\\" ," +
                    "\"\'\"oozieJobId\"\'\":\\\"" + oozieJobId + "\\\"," +
                    "\"\'\"stat_date\"\'\":\\\"${param}\\\" ," +
                    "\"\'\"isCorJob\"\'\":\\\"" + true + "\\\"}\" \"" + oozieConfig.getRest_url() + "/job_begin\"`\n"
                    +"echo ${depOOzieId}\nif [[ ! -n \"${depOOzieId}\" ]]; then\n\texit 10\nfi\nif [[ ${depOOzieId} == '-1' ]]; then\n\texit 10\nfi\necho \"depOOzieId:$depOOzieId\"";

        }
    }

    private List<List<JobNodeInfoDto>> listAllJobNodeList(List<JobLinkDto> edges, Set<Integer> olderNodeIds) {
        //找出所有的连线（包括虚拟的开始节点）
        List<List<Integer>> lists = listLinkedList(edges, olderNodeIds);
        if (CollectionUtil.isEmpty(lists)) {
            return null;
        }

        Set<Integer> nodeIds = new HashSet<>();
        for (List<Integer> list : lists) {
            for (int i = 1; i < list.size(); i++) {
                nodeIds.add(list.get(i));
            }
        }
        //查询到所节点的数据和对应的配置信息
        List<JobNodeInfoDto> jobNodeInfoDtos = jobNodeInfoService.listAllNodeList(nodeIds);
        if (CollectionUtil.isEmpty(jobNodeInfoDtos)) {
            return null;
        }
        //直接生成对应的表达式
        List<List<JobNodeInfoDto>> allLinkedDtoList = new ArrayList<>();
        JobNodeInfoDto startJobNodeInfoDto = new JobNodeInfoDto();
        startJobNodeInfoDto.setWorkFlowNodeName(JobType.ObjOf(0).getEnName() + "_0");
        for (List<Integer> list : lists) {
            List<JobNodeInfoDto> list1 = new ArrayList<>();
            list1.add(startJobNodeInfoDto);
            for (int i = 1; i < list.size(); i++) {
                int finalI = i;
                List<JobNodeInfoDto> collect = jobNodeInfoDtos.stream().filter(jobNodeInfoDto -> Objects.equals(list.get(finalI), jobNodeInfoDto.getId())).collect(Collectors.toList());
                list1.add(collect.get(0));
            }
            allLinkedDtoList.add(list1);
        }
        return allLinkedDtoList;

    }

    /***
     * 验证节点连线-----自己手动添加一个开始节点和一个结束节点，再来验证连线和节点是否正确
     * @param edges  节点连线
     * @param nodeIds  所有节点对应的id集合
     */
    private void cheackLinked(List<JobLinkDto> edges, Set<Integer> nodeIds) {

        if (CollectionUtil.isEmpty(nodeIds)) {
            throw new RRException("请配置节点!");
        }

        //连线和节点匹配
        if(CollectionUtil.isNotEmpty(edges)){
            for (JobLinkDto edge : edges) {
                if (!nodeIds.contains(edge.getSrcNodeId()) || !nodeIds.contains(edge.getDstNodeId())) {
                    throw new RRException("连线和节点不匹配!");
                }
            }
        }


        //连线是否规范
        //1.得到所有的连线分支
        //2.组装开始节点和结束节点
        //3.判断连线是否正确
        List<List<Integer>> linkList = listLinkedList(edges, nodeIds);

        //验证连线情况
        if (null == linkList || linkList.isEmpty()) {
            throw new RRException(DataDevelopmentBizExceptionEnum.JOB_NODE_NULL.getMessage());
        }
        //开始节点下面的连线情况
        List<List<Integer>> beginList = linkList.stream().filter(dataList -> dataList.size() > 1).collect(Collectors.toList());
        if (beginList.isEmpty()) {
            throw new RRException(DataDevelopmentBizExceptionEnum.START_NODE_WITHOUT_NEXT_NODE.getMessage());
        }
        //验证连线是否存在循环引用的连线
        checkLinkToLink(linkList);

    }
    private List<List<String>> listLinkedList1(List<JSONObject> edges, Set<String> nodeIds) {
        Set<String> startNodeIds = new HashSet<>();
        if (CollectionUtil.isEmpty(edges)) {
            startNodeIds.addAll(nodeIds);
        }else{
            List<String> allNodeIds = new ArrayList<>();
            Set<String> srcAllNodeIds = new HashSet<>();
            Set<String> dstAllNodeIds = new HashSet<>();

            for (JSONObject edge : edges) {
                srcAllNodeIds.add(edge.getString("srcNodeId"));
                dstAllNodeIds.add(edge.getString("dstNodeId"));
                allNodeIds.add(edge.getString("srcNodeId"));
                allNodeIds.add(edge.getString("dstNodeId"));
            }
            //获取起节点
            for (String srcAllNodeId : srcAllNodeIds) {
                if (!dstAllNodeIds.contains(srcAllNodeId)) {
                    startNodeIds.add(srcAllNodeId);
                }
            }
            //如果，没有连线也必须算起节点
            for (String nodeId : nodeIds) {
                if (!allNodeIds.contains(nodeId)) {
                    startNodeIds.add(nodeId);
                }
            }
            if (CollectionUtil.isEmpty(startNodeIds)) {
                throw new RRException("节点是存在循环引用!");
            }
        }

        //组装开始节点
        List<JSONObject> initList = new ArrayList<>();
        initStartNode1(edges, startNodeIds);
        edges.addAll(initList);

        //递归获取到连线
        //从开始节点开始，依次查找到对应的节点信息
        List<List<String>> beginLinkList = new ArrayList<>();
        List<String> beginStr = new ArrayList<>();
        beginStr.add("0");
        beginLinkList.add(beginStr);
        List<List<String>> linkList = reducLinkNodeList1(edges, beginStr, beginLinkList);
        return linkList;
    }

    private List<List<String>> reducLinkNodeList1(List<JSONObject> edges, List<String> satrtList, List<List<String>> inputLinkList) {
        List<List<String>> tempList = new ArrayList<>();
        if (CollectionUtil.isEmpty(edges) || CollectionUtil.isEmpty(satrtList)) {
            return inputLinkList;
        }
        Set<String> beginSet = new HashSet<>();
        for (String start : satrtList) {
            List<JSONObject> collect = edges.stream().filter(jobLinkDto1 -> Objects.equals(start, jobLinkDto1.getString("srcNodeId"))).collect(Collectors.toList());
            //获取到对应的已经保存过的数据集合
            List<List<String>> pingList = inputLinkList.stream().filter(dataList -> Objects.equals(dataList.get(dataList.size() - 1), start)).collect(Collectors.toList());
            for (List<String> integers : pingList) {
                if (collect.isEmpty()) {
                    //需要判断的节点连线
                    beginSet.add(start);

                    //重新添加数据
                    tempList.add(integers);
                }
                for (JSONObject jobLinkDto : collect) {
                    List<String> keyTempList = new ArrayList<>(integers);
                    keyTempList.add(jobLinkDto.getString("dstNodeId"));
                    //重新添加数据
                    tempList.add(keyTempList);

                    //新的这个是需要去重的
                    beginSet.add(jobLinkDto.getString("dstNodeId"));
                }
            }
        }
        List<String> beginNodeTempList = new ArrayList<>(beginSet);

        //找不到依赖也应该对应的结束循环
        if (tempList.isEmpty()) {
            return inputLinkList;
        }

        //当输入的和输出的结果一样的时候，也可以结束
        if (tempList.size() == inputLinkList.size() && tempList.containsAll(inputLinkList)) {
            return inputLinkList;
        }

        //把使用过的节点 。应该以结束标准来表示对应的结束循环,其中最大的一个连线个数达到了这个连线集合的总个数，则直接跳出循环
        int maxCountSize = tempList.get(0).size();
        for (int i = 1; i < tempList.size(); i++) {
            if (tempList.get(i).size() > maxCountSize) {
                maxCountSize = tempList.get(i).size();
            }
        }
        //这里必须+1
        if (maxCountSize >= edges.size() + 1) {
            return tempList;
        }

        //递归调用
        return reducLinkNodeList1(edges, beginNodeTempList, tempList);
    }

    @Override
    public List<List<Integer>> listLinkedList(List<JobLinkDto> edges, Set<Integer> nodeIds) {
        Set<Integer> startNodeIds = new HashSet<>();
        if (CollectionUtil.isEmpty(edges)) {
            edges = new ArrayList<>();
            startNodeIds.addAll(nodeIds);
        }else{
            List<Integer> allNodeIds = new ArrayList<>();
            Set<Integer> srcAllNodeIds = new HashSet<>();
            Set<Integer> dstAllNodeIds = new HashSet<>();
            for (JobLinkDto edge : edges) {
                srcAllNodeIds.add(edge.getSrcNodeId());
                dstAllNodeIds.add(edge.getDstNodeId());
                allNodeIds.add(edge.getSrcNodeId());
                allNodeIds.add(edge.getDstNodeId());
            }
            //获取起节点
            for (Integer srcAllNodeId : srcAllNodeIds) {
                if (!dstAllNodeIds.contains(srcAllNodeId)) {
                    startNodeIds.add(srcAllNodeId);
                }
            }
            //如果，没有连线也必须算起节点
            for (Integer nodeId : nodeIds) {
                if (!allNodeIds.contains(nodeId)) {
                    startNodeIds.add(nodeId);
                }
            }
            if (CollectionUtil.isEmpty(startNodeIds)) {
                throw new RRException("节点是存在循环引用!");
            }
        }
        //组装开始节点
        List<JobLinkDto> initList = new ArrayList<>();
        initStartNode(edges, startNodeIds);
        edges.addAll(initList);

        //递归获取到连线
        //从开始节点开始，依次查找到对应的节点信息
        List<List<Integer>> beginLinkList = new ArrayList<>();
        List<Integer> beginStr = new ArrayList<>();
        beginStr.add(0);
        beginLinkList.add(beginStr);
        List<List<Integer>> linkList = reducLinkNodeList(edges, beginStr, beginLinkList);
        return linkList;
    }

    private void checkLinkToLink(List<List<Integer>> linkList) {
        if (CollectionUtil.isEmpty(linkList)) {
            return;
        }
        //如果节点中有重复的key,那么肯定是循环引用
        for (List<Integer> strings : linkList) {
            for (int i = 0; i < strings.size() - 1; i++) {
                for (int j = i + 1; j < strings.size(); j++) {
                    if (Objects.equals(strings.get(i), strings.get(j))) {
                        String errorFromMsg = "", errorToMsg = "";
                        throw new RRException(String.format("存在循环指向%s任务节点到%s任务节点！", errorFromMsg, errorToMsg));
                    }
                }
            }
        }
    }


    private List<List<Integer>> reducLinkNodeList(List<JobLinkDto> edges, List<Integer> satrtList, List<List<Integer>> inputLinkList) {
        List<List<Integer>> tempList = new ArrayList<>();
        if (CollectionUtil.isEmpty(edges) || CollectionUtil.isEmpty(satrtList)) {
            return inputLinkList;
        }
        Set<Integer> beginSet = new HashSet<>();
        for (Integer start : satrtList) {
            List<JobLinkDto> collect = edges.stream().filter(jobLinkDto1 -> Objects.equals(start, jobLinkDto1.getSrcNodeId())).collect(Collectors.toList());
            //获取到对应的已经保存过的数据集合
            List<List<Integer>> pingList = inputLinkList.stream().filter(dataList -> Objects.equals(dataList.get(dataList.size() - 1), start)).collect(Collectors.toList());
            for (List<Integer> integers : pingList) {
                if (collect.isEmpty()) {
                    //需要判断的节点连线
                    beginSet.add(start);

                    //重新添加数据
                    tempList.add(integers);
                }
                for (JobLinkDto jobLinkDto : collect) {
                    List<Integer> keyTempList = new ArrayList<>(integers);
                    keyTempList.add(jobLinkDto.getDstNodeId());
                    //重新添加数据
                    tempList.add(keyTempList);

                    //新的这个是需要去重的
                    beginSet.add(jobLinkDto.getDstNodeId());
                }
            }
        }
        List<Integer> beginNodeTempList = new ArrayList<>(beginSet);

        //找不到依赖也应该对应的结束循环
        if (tempList.isEmpty()) {
            return inputLinkList;
        }

        //当输入的和输出的结果一样的时候，也可以结束
        if (tempList.size() == inputLinkList.size() && tempList.containsAll(inputLinkList)) {
            return inputLinkList;
        }

        //把使用过的节点 。应该以结束标准来表示对应的结束循环,其中最大的一个连线个数达到了这个连线集合的总个数，则直接跳出循环
        int maxCountSize = tempList.get(0).size();
        for (int i = 1; i < tempList.size(); i++) {
            if (tempList.get(i).size() > maxCountSize) {
                maxCountSize = tempList.get(i).size();
            }
        }
        //这里必须+1
        if (maxCountSize >= edges.size() + 1) {
            return tempList;
        }

        //递归调用
        return reducLinkNodeList(edges, beginNodeTempList, tempList);
    }

    private void initStartNode(List<JobLinkDto> initList, Set<Integer> startNodeIds) {
        for (Integer satrtNode : startNodeIds) {
            JobLinkDto jobLinkDto = new JobLinkDto();
            jobLinkDto.setId(0);
            jobLinkDto.setName("开始（虚拟）节点");
            jobLinkDto.setDstNodeId(satrtNode);
            jobLinkDto.setSrcNodeId(0);
            initList.add(jobLinkDto);
        }
    }

    private void initStartNode1(List<JSONObject> initList, Set<String> startNodeIds) {
        for (String satrtNode : startNodeIds) {
            JSONObject jobLinkDto = new JSONObject();
            jobLinkDto.put("dstNodeId",satrtNode);
            jobLinkDto.put("srcNodeId","0");
            initList.add(jobLinkDto);
        }
    }


    /***
     * 1.路由条件  多父节点路由,暂时不支持！
     * 2.错误路由  失败路由节点，暂时不支持有下游依赖节点   和 当前节点的父节点存在多个失败的路由条件，暂时不支持
     * @param nodes
     * @param edges
     * @param jobId
     */
    private void cheackJobNodeLayConfig(List<JobNodeInfoDto> nodes, List<JobLinkDto> edges, Long jobId) {
        if(CollectionUtil.isEmpty(nodes)){
            return;
        }
        if(CollectionUtil.isNotEmpty(edges)){
            Set<Integer> nodeIds = new HashSet<>();
            for (JobLinkDto node : edges) {
                nodeIds.add(node.getDstNodeId());
                nodeIds.add(node.getSrcNodeId());
            }
            //查询到这个工作流以前配置的所有的路由判断数据
            List<JobNodeLayConfig> jobNodeLayConfigs = jobNodeLayConfigMapper.selectList(new QueryWrapper<JobNodeLayConfig>()
                    .eq("job_id", jobId));
            if(CollectionUtil.isNotEmpty(jobNodeLayConfigs)){
                Set<JobNodeLayConfig> countList = jobNodeLayConfigs.stream().filter(o -> (Objects.equals(o.getConfigType(), RouteTypeEnum.COUNT.getId()) && null != o.getOutputTableId())).collect(Collectors.toSet());
                Set<JobNodeLayConfig> errList = jobNodeLayConfigs.stream().filter(o -> (Objects.equals(o.getConfigType(), RouteTypeEnum.STATUS.getId()) && Objects.equals(o.getConditionValue(), "1"))).collect(Collectors.toSet());

                //如果是路由条件 多父节点路由,暂时不支持！
                if(CollectionUtil.isNotEmpty(countList)){
                    for (JobNodeLayConfig jobNodeLayConfig : countList) {
                        if(nodeIds.contains(jobNodeLayConfig.getCurrentNodeId())){
                            List<JobLinkDto> collects = edges.stream().filter(o -> Objects.equals(o.getDstNodeId(), jobNodeLayConfig.getCurrentNodeId())).collect(Collectors.toList());
                            if(CollectionUtil.isNotEmpty(collects) && collects.size() > 1){
                                StringBuilder sb = new StringBuilder();
                                JobNodeInfoDto jobNodeInfoDto = nodes.stream().filter(o -> Objects.equals(o.getId(), jobNodeLayConfig.getCurrentNodeId())).findAny().orElse(null);
                                String jobNodeName = jobNodeInfoDto == null ? "" : jobNodeInfoDto.getName();
                                sb.append("有配置路由条件,暂时不支持多父节点路由节点路由到当前节点：【"+jobNodeName+"】！请改变连线或者修改路由条件判断！");
                                throw new RRException(sb.toString());
                            }
                        }
                    }
                }
                //如果是错误路由
                if(CollectionUtil.isNotEmpty(errList)){
                    for (JobNodeLayConfig jobNodeLayConfig : errList) {
                        if(nodeIds.contains(jobNodeLayConfig.getCurrentNodeId())){
                            //失败路由节点，暂时不支持有下游依赖节点
                            List<JobLinkDto> collects = edges.stream().filter(o -> Objects.equals(o.getSrcNodeId(), jobNodeLayConfig.getCurrentNodeId())).collect(Collectors.toList());
                            if(CollectionUtil.isNotEmpty(collects) && collects.size() >= 1){
                                StringBuilder sb = new StringBuilder();
                                JobNodeInfoDto jobNodeInfoDto = nodes.stream().filter(o -> Objects.equals(o.getId(), jobNodeLayConfig.getCurrentNodeId())).findAny().orElse(null);
                                String jobNodeName = jobNodeInfoDto == null ? "" : jobNodeInfoDto.getName();
                                sb.append("有配置失败路由,暂时不支持失败路由到当前节点：【"+jobNodeName+"】,还配置下游依赖节点！请改变连线或者修改路由条件判断！");
                                throw new RRException(sb.toString());
                            }
                            //失败路由的节点暂时不支持配置其它的路由方式
                            cheackErrorNode(jobNodeLayConfig,nodes,edges,jobNodeLayConfigs);

                            //当前节点的父节点存在多个失败的路由条件，暂时不支持
                            Set<JobLinkDto> collect = edges.stream().filter(o -> Objects.equals(o.getDstNodeId(), jobNodeLayConfig.getCurrentNodeId())).collect(Collectors.toSet());
                            if(CollectionUtil.isNotEmpty(collect)){
                                Set<Integer> pids = collect.stream().map(JobLinkDto::getSrcNodeId).collect(Collectors.toSet());
                                if(CollectionUtil.isNotEmpty(pids)){
                                    for (Integer pid : pids) {
                                        Set<JobNodeLayConfig> collect1 = errList.stream().filter(o -> Objects.equals(pid, o.getUpNodeId())).collect(Collectors.toSet());
                                        if(CollectionUtil.isNotEmpty(collect1)){
                                            Set<Integer> collect2 = collect1.stream().map(JobNodeLayConfig::getCurrentNodeId).collect(Collectors.toSet());
                                            collect2.remove(jobNodeLayConfig.getCurrentNodeId());
                                            if(CollectionUtil.isNotEmpty(collect2)){
                                                StringBuilder sb = new StringBuilder();
                                                JobNodeInfoDto jobNodeInfoDto = nodes.stream().filter(o -> Objects.equals(o.getId(), jobNodeLayConfig.getCurrentNodeId())).findAny().orElse(null);
                                                String jobNodeName = jobNodeInfoDto == null ? "" : jobNodeInfoDto.getName();
                                                JobNodeInfoDto pjobNodeInfoDto = nodes.stream().filter(o -> Objects.equals(o.getId(), pid)).findAny().orElse(null);
                                                String pjobNodeName = pjobNodeInfoDto == null ? "" : pjobNodeInfoDto.getName();
                                                sb.append("有配置失败路由,暂时不支持父节点：【"+pjobNodeName+"】失败路由到多个下游节点：【"+jobNodeName+"】");
                                                for (Integer integer : collect2) {
                                                    JobNodeInfoDto tempjobNodeInfoDto = nodes.stream().filter(o -> Objects.equals(o.getId(), integer)).findAny().orElse(null);
                                                    String tempjobNodeName = tempjobNodeInfoDto == null ? "" : tempjobNodeInfoDto.getName();
                                                    sb.append(",【").append(tempjobNodeName).append("】");
                                                }
                                                sb.append(",请改变连线或者修改路由条件判断！");
                                                throw new RRException(sb.toString());
                                            }
                                        }
                                    }
                                }
                            }
                        }
                    }
                }
            }
        }
    }

    private void cheackErrorNode(JobNodeLayConfig jobNodeLayConfig, List<JobNodeInfoDto> nodes, List<JobLinkDto> edges, List<JobNodeLayConfig> jobNodeLayConfigs) {
        if(CollectionUtil.isEmpty(edges)){
            return;
        }
        if(CollectionUtil.isEmpty(nodes)){
            return;
        }
        Set<Integer> upIds = edges.stream().filter(o -> Objects.equals(o.getDstNodeId(), jobNodeLayConfig.getCurrentNodeId())).map(JobLinkDto::getSrcNodeId).collect(Collectors.toSet());
        if(CollectionUtil.isNotEmpty(upIds)){
            if(upIds.size() > 1){
                Set<Integer> upNodeIds = new HashSet<>();
                for (Integer upId : upIds) {
                    if(CollectionUtil.isNotEmpty(jobNodeLayConfigs)){
                        JobNodeLayConfig jobNodeLayConfig1 = jobNodeLayConfigs.stream().filter(o -> (Objects.equals(o.getUpNodeId(), upId) && Objects.equals(o.getCurrentNodeId(), jobNodeLayConfig.getCurrentNodeId()))).findAny().orElse(null);
                        if(null != jobNodeLayConfig1){
                            if(!Objects.equals(jobNodeLayConfig1.getConfigType(), RouteTypeEnum.STATUS.getId())
                                    || (Objects.equals(jobNodeLayConfig1.getConfigType(), RouteTypeEnum.STATUS.getId()) && !Objects.equals(jobNodeLayConfig1.getConditionValue(), "1"))){
                                upNodeIds.add(upId);
                            }
                        }else{
                            //可能新增的默认为成功的连线
                            upNodeIds.add(upId);
                        }
                    }
                }
                if(CollectionUtil.isNotEmpty(upNodeIds)){
                    JobNodeInfoDto jobNodeInfoDto = nodes.stream().filter(o -> Objects.equals(o.getId(), jobNodeLayConfig.getCurrentNodeId())).findAny().orElse(null);
                    String jobNodeName = jobNodeInfoDto == null ? "" : jobNodeInfoDto.getName();
                    throw new RRException("有配置失败路由,暂时不支持失败路由和其它路由方式(条件、成功)路由到当前节点：【"+jobNodeName+"】,请改变连线或者修改路由条件判断！");
                }
            }
        }
    }

    /***
     * 通过连线，改变以前配置的依赖关系
     * @param nodes       节点集合
     * @param edges       连线集合
     * @param jobId  jobId
     */
    private void changeJobNodeLayConfig(List<JobNodeInfoDto> nodes, List<JobLinkDto> edges, Long jobId) {
        if (CollectionUtil.isEmpty(nodes) || CollectionUtil.isEmpty(edges)) {
            //清空所有的这个job对应的依赖
            jobNodeLayConfigMapper.delete(new QueryWrapper<JobNodeLayConfig>().eq("job_id", jobId));
            return;
        }
        //如果原来有的连线，现在还是有的话，则不删除其依赖配置。
        //其它的情况只添加上下游连线，但是，还是得删除对应的依赖条件配置
        List<JobNodeLayConfig> olderJobNodeLayConfigs = new ArrayList<>();
        List<JobNodeLayConfig> addJobNodeLayConfigs = new ArrayList<>();
        for (JobLinkDto linkDto : edges) {
            List<JobNodeLayConfig> jobNodeLayConfigs = jobNodeLayConfigMapper.selectList(new QueryWrapper<JobNodeLayConfig>()
                    .eq("up_node_id", linkDto.getSrcNodeId())
                    .eq("current_node_id", linkDto.getDstNodeId())
                    .eq("job_id", jobId));
            if (CollectionUtil.isNotEmpty(jobNodeLayConfigs)) {
                olderJobNodeLayConfigs.addAll(jobNodeLayConfigs);
            } else {
                JobNodeInfo jobNodeInfo = jobNodeInfoMapper.selectById(linkDto.getDstNodeId());
                if (null == jobNodeInfo) {
                    throw new RRException("节点不存在，id:" + linkDto.getDstNodeId());
                }
                JobNodeLayConfig jobNodeLayConfig = new JobNodeLayConfig();
                jobNodeLayConfig.setJobId(jobId.intValue());
                jobNodeLayConfig.setConfigType(RouteTypeEnum.STATUS.getId());
                jobNodeLayConfig.setConditionNamme(RouteTypeEnum.STATUS.getName());
                jobNodeLayConfig.setOperationNamme("=");
                jobNodeLayConfig.setConditionValue("0");
                jobNodeLayConfig.setStatesNamme(RouteTypeEnum.STATUS.getName());
                jobNodeLayConfig.setStatesValue("成功");
                jobNodeLayConfig.setUpNodeId(linkDto.getSrcNodeId());
                jobNodeLayConfig.setCurrentNodeId(linkDto.getDstNodeId());
                jobNodeLayConfig.setCreateTime(DateTime.now());
                jobNodeLayConfig.setDutyUser(jobNodeInfo.getCreatePer());
                addJobNodeLayConfigs.add(jobNodeLayConfig);
            }
        }
        //先删除，然后新增
        jobNodeLayConfigMapper.delete(new QueryWrapper<JobNodeLayConfig>().eq("job_id", jobId));
        if (CollectionUtil.isNotEmpty(olderJobNodeLayConfigs)) {
            jobNodeLayConfigMapper.batchInsert(olderJobNodeLayConfigs);
        }
        if (CollectionUtil.isNotEmpty(addJobNodeLayConfigs)) {
            jobNodeLayConfigMapper.batchInsert(addJobNodeLayConfigs);
        }

    }

    @Override
    public List<JobInfoDto> listJobList(WorkMenuDto workMenuDto) {
        //查询出来位置不为null的，则一定是有拖入画布的
        if (null == workMenuDto || workMenuDto.getId() == null) {
            throw new RRException(DataDevelopmentBizExceptionEnum.PARAM_ERROR.getMessage());
        }
        WorkMenu workMenu = workMenuMapper.selectById(workMenuDto.getId());
        if (null == workMenu) {
            throw new RRException("默认作业流依赖管理不存在!");
        }
        if (WorkFlowMenuType.WORKFLOW_MGT_FIELD.getCode() != workMenu.getWorkType()) {
            throw new RRException("上级目录不是默认作业流依赖管理类型!");
        }
        List<WorkMenuJobRel> workMenuJobRels = workMenuJobRelMapper.selectList(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 2).eq("work_menu_id", workMenuDto.getId()));
        Set<Integer> jobIds = new HashSet<>();
        for (WorkMenuJobRel workMenuJobRel : workMenuJobRels) {
            jobIds.add(workMenuJobRel.getWorkFlowId().intValue());
        }
        if (CollectionUtil.isEmpty(jobIds)) {
            return null;
        }
        List<JobInfoDto> listJobInfoDtos = jobInfoMapper.selectListByIds(jobIds);
        if (CollectionUtil.isNotEmpty(listJobInfoDtos)) {
            //查询到对应的任务数据
            for (JobInfoDto listJobInfoDto : listJobInfoDtos) {
                WorkMenuJobRel workMenuJobRel = workMenuJobRelMapper.selectOne(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 1).eq("work_flow_id", listJobInfoDto.getId()));
                if(null != workMenuJobRel){
                    listJobInfoDto.setRealId(workMenuJobRel.getWorkMenuId());
                }
            }
            return listJobInfoDtos;
        }
        return null;
    }

    @Override
    public void updateWorkFlow(JobLinkAndJobDto jobLinkAndJobDto, Boolean aTrue) {

        if (null == jobLinkAndJobDto.getProcId()) {
            throw new RRException("工程id不能为空!");
        }
        WorkMenu workMenu = workMenuMapper.selectById(jobLinkAndJobDto.getWorkFlowMenuId());
        if (null == workMenu) {
            throw new RRException("默认作业流依赖管理不存在!");
        }
        if (WorkFlowMenuType.WORKFLOW_MGT_FIELD.getCode() != workMenu.getWorkType()) {
            throw new RRException("上级目录不是默认作业流依赖管理类型!");
        }
        //查询其工作流依赖数据信息
        WorkFlowDep workFlowDep = workFlowDepMapper.selectOne(new QueryWrapper<WorkFlowDep>().eq("work_menu_id", workMenu.getId()));
        if (null == workFlowDep) {
            throw new RRException("默认工作流依赖不存在!");
        }
        //判断这个工作流依赖的运行状态，如果是正在运行中，则不能修改
        if (Objects.equals(1, workFlowDep.getJobLayStatus())) {
            throw new RRException("默认工作流依赖不能修改,请先禁用工作流依赖!");
        }

        //验证对应的用户是否可以操作这个工程
        cheackProcId(jobLinkAndJobDto.getProcId(), ShiroUtils.getUserId(), true, workMenu.getCreatePer(),"admin",jobLinkAndJobDto.getIsImport());

        //验证这些job任务没有配置过正常调度并且都有配置作业流
        if(aTrue){
            cheackJobSchulde(jobLinkAndJobDto.getNodes());
        }
        cheackHaveJobLinked(jobLinkAndJobDto.getNodes());
        cheackJobCount(jobLinkAndJobDto.getNodes(), workMenu.getId());
        cheackJobIsWorkFlowMenu(jobLinkAndJobDto.getNodes(), workMenu.getRootId());
        cheackExpressionNumber(jobLinkAndJobDto.getJobNodeConfig());


        if(CollectionUtil.isEmpty(jobLinkAndJobDto.getNodes())){
            //清空数据
            List<WorkMenuJobRel> olserWorkMenuJobRels = workMenuJobRelMapper.selectList(new QueryWrapper<WorkMenuJobRel>().eq("work_menu_id", jobLinkAndJobDto.getWorkFlowMenuId()).eq("work_type", 2));
            if (CollectionUtil.isNotEmpty(olserWorkMenuJobRels)) {
                workMenuJobRelMapper.delete(new QueryWrapper<WorkMenuJobRel>().eq("work_menu_id", jobLinkAndJobDto.getWorkFlowMenuId()).eq("work_type", 2));
                Set<Integer> jobIds = new HashSet<>();
                for (WorkMenuJobRel workMenuJobRel : olserWorkMenuJobRels) {
                    jobIds.add(workMenuJobRel.getWorkFlowId().intValue());
                }
                jobInfoMapper.clearPosOthers(jobIds);
            }
            jobLinkMapper.delete(new QueryWrapper<JobLink>().eq("linke_type", 1).eq("menu_id", jobLinkAndJobDto.getWorkFlowMenuId()));
            //保存基本属性配置
            saveScheduleData(jobLinkAndJobDto.getJobNodeConfig(), jobLinkAndJobDto.getWorkFlowMenuId(), true);

            return;
        }

        //验证工作流节点和连线信息
        Set<Integer> nodeIds = new HashSet<>();
        for (JobInfoDto jobInfoDto : jobLinkAndJobDto.getNodes()) {
            nodeIds.add(jobInfoDto.getId());
        }
        cheackLinked(jobLinkAndJobDto.getEdges(), nodeIds);


        //保存节点和连线信息
        //查询到所有的节点并且删除
        List<WorkMenuJobRel> olserWorkMenuJobRels = workMenuJobRelMapper.selectList(new QueryWrapper<WorkMenuJobRel>().eq("work_menu_id", jobLinkAndJobDto.getWorkFlowMenuId()).eq("work_type", 2));
        if (CollectionUtil.isNotEmpty(olserWorkMenuJobRels)) {
            workMenuJobRelMapper.delete(new QueryWrapper<WorkMenuJobRel>().eq("work_menu_id", jobLinkAndJobDto.getWorkFlowMenuId()).eq("work_type", 2));
            Set<Integer> jobIds = new HashSet<>();
            for (WorkMenuJobRel workMenuJobRel : olserWorkMenuJobRels) {
                jobIds.add(workMenuJobRel.getWorkFlowId().intValue());
            }
            jobInfoMapper.clearPosOthers(jobIds);
        }
        if (CollectionUtil.isNotEmpty(jobLinkAndJobDto.getNodes())) {
            //更新节点的位置
            jobInfoMapper.batchUpdate(jobLinkAndJobDto.getNodes());
            List<WorkMenuJobRel> workMenuJobRels = new ArrayList<>();
            for (JobInfoDto jobInfoDto : jobLinkAndJobDto.getNodes()) {
                WorkMenuJobRel workMenuJobRel = new WorkMenuJobRel();
                workMenuJobRel.setProcId(jobLinkAndJobDto.getProcId());
                workMenuJobRel.setWorkType(2);
                workMenuJobRel.setWorkMenuId(jobLinkAndJobDto.getWorkFlowMenuId());
                workMenuJobRel.setWorkFlowId(Long.valueOf(jobInfoDto.getId()));
                workMenuJobRel.setCreateTime(LocalDateTime.now());
                workMenuJobRels.add(workMenuJobRel);
            }
            workMenuJobRelMapper.batchInsert(workMenuJobRels);
        }


        jobLinkMapper.delete(new QueryWrapper<JobLink>().eq("linke_type", 1).eq("menu_id", jobLinkAndJobDto.getWorkFlowMenuId()));
        //重新获取去掉开始节点
        List<JobLinkDto> insertDataList = new ArrayList<>();
        if(CollectionUtil.isNotEmpty(jobLinkAndJobDto.getEdges())){
            jobLinkAndJobDto.getEdges().forEach(jobLinkDto -> {
                if (0 != jobLinkDto.getSrcNodeId()) {
                    JobLinkDto jobLinkDto1 = new JobLinkDto();
                    BeanUtils.copyProperties(jobLinkDto, jobLinkDto1);
                    jobLinkDto1.setCreatePer(ShiroUtils.getUserId().intValue());
                    jobLinkDto1.setCreateTime(DateTime.now());
                    jobLinkDto1.setLinkeType(1);
                    jobLinkDto1.setMenuId(jobLinkAndJobDto.getWorkFlowMenuId());
                    insertDataList.add(jobLinkDto1);
                }
            });
        }
        if (CollectionUtil.isNotEmpty(insertDataList)) {
            jobLinkMapper.batchInsert(insertDataList);
        }

        //保存基本属性配置
        saveScheduleData(jobLinkAndJobDto.getJobNodeConfig(), jobLinkAndJobDto.getWorkFlowMenuId(), true);

        //保存每个节点对应的shell.sh文件。回调启动任务和查询任务的运行状态
        saveAllJobShell(jobLinkAndJobDto.getProcId(), workFlowDep.getId(), jobLinkAndJobDto.getNodes());

    }

    private void cheackJobIsWorkFlowMenu(List<JobInfoDto> nodes, Long rootId) {
        //验证这些job任务是否都在同一个根目录下
        if (CollectionUtil.isEmpty(nodes)) {
            return;
        }
        List<Integer> jobIds = new ArrayList<>();
        List<WorkMenuJobRel> restList = workMenuJobRelMapper.selectListJobByRootId(rootId);
        for (WorkMenuJobRel workMenuJobRel : restList) {
            jobIds.add(workMenuJobRel.getWorkFlowId().intValue());
        }
        for (JobInfoDto node : nodes) {
            if (!jobIds.contains(node.getId())) {
                throw new RRException("工作流依赖引用了不是当前根目录下的工作流:【" + node.getName() + "】");
            }
        }
    }

    private void cheackJobCount(List<JobInfoDto> nodes, Long menuId) {
        //一个工作流依赖，针对于所有的工作流，只能配置一个
        if(CollectionUtil.isEmpty(nodes)){
            return;
        }
        for (JobInfoDto node : nodes) {
            List<WorkMenuJobRel> workMenuJobRels = workMenuJobRelMapper.selectList(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 2).eq("work_flow_id", node.getId()).ne("work_menu_id", menuId));
            if (CollectionUtil.isNotEmpty(workMenuJobRels)) {
                JobInfo jobInfo = jobInfoMapper.selectById(node.getId());
                throw new RRException("【"+jobInfo.getName()+"】工作流已被其它作业流依赖引用! ");
            }
        }
    }

    /***
     * 将job对应的shell保存到对应的
     * @param procId         工程id
     * @param id             作业流依赖id
     * @param nodes          作业流依赖的job的信息
     */
    private void saveAllJobShell(Integer procId, Long id, List<JobInfoDto> nodes) {
        if (CollectionUtil.isEmpty(nodes)) {
            return;
        }
        // 将shell保存至hdfs
        String baseDst = "workflow-app/" + WorkFlowType.WORKFLOW_LAY.getCode() + "/" + procId + "/" + id + "/mid_";
        for (JobInfoDto node : nodes) {
            String dst = baseDst + node.getId() + ".sh";
            jobNodeInfoService.uploadJobLayShell(procId, node.getId(), dst, oozieConfig);
        }
    }

    private void cheackHaveJobLinked(List<JobInfoDto> nodes) {
        if (CollectionUtil.isEmpty(nodes)) {
            return;
        }
        for (JobInfoDto node : nodes) {
            JobInfo jobInfo = jobInfoMapper.selectById(node.getId());
            if (null == jobInfo) {
                throw new RRException("不存在工作流！jobId:" + node.getId());
            }
            //查询到对应的菜单
            WorkMenuJobRel workMenuJobRel = workMenuJobRelMapper.selectOne(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 1).eq("work_flow_id", node.getId()));
            if (null == workMenuJobRel) {
                throw new RRException("不存在【"+jobInfo.getName()+"】工作流！");
            }
            //查询对应的连线---只查询对应的节点
            List<JobNodeInfo> jobNodeInfos = jobNodeInfoMapper.selectList(new QueryWrapper<JobNodeInfo>().eq("job_id", node.getId()).isNotNull("pos_x").isNotNull("pos_y"));
            if (CollectionUtil.isEmpty(jobNodeInfos)) {
                throw new RRException( jobInfo.getName()+"工作流未配置节点！" );
            }
        }

    }

    private void cheackJobSchulde(List<JobInfoDto> nodes) {
        if (CollectionUtil.isEmpty(nodes)) {
            return;
        }
        //添加验证重复的验证
        Set<Integer> lists = new HashSet<>();
        for (JobInfoDto node : nodes) {
            lists.add(node.getId());
        }
        if(lists.size() != nodes.size()){
            throw new RRException("依赖里面存在相同的作业流！");
        }

        for (JobInfoDto node : nodes) {
            //查询到对应的菜单id
            WorkMenuJobRel workMenuJobRel = workMenuJobRelMapper.selectOne(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 1).eq("work_flow_id", node.getId()));
            if(null == workMenuJobRel){
                throw new RRException("任务对应的菜单映射不存在！");
            }
            JobNodeConf jobNodeConf = jobNodeConfMapper.selectOne(new QueryWrapper<JobNodeConf>().eq("`key`", "schedule_type").eq("job_node_id", workMenuJobRel.getWorkMenuId()).eq("job_type", 1));
            if (null != jobNodeConf && "1".equals(jobNodeConf.getValue())) {
                JobInfo jobInfo = jobInfoMapper.selectById(node.getId());
                throw new RRException("请先把【"+jobInfo.getName()+"】作业流调度方式改为暂停!");
            }
        }

    }


    /***
     * 生成对应的workFlow依赖任务
     * @param edges  节点连线
     * @param jobNodeConfig  作业流的配置属性
     * @param workFlowDepId          任务流依赖id
     */
    private void generatedAutoWorkFlowLayXml(List<JobLinkDto> edges, Set<Integer> ids, JobNodeConfig jobNodeConfig, Long workFlowDepId, Integer procId) throws Exception {

        WorkFlowDep workFlowDep = workFlowDepMapper.selectById(workFlowDepId);
        if (null == workFlowDep) {
            throw new RRException("工作流依赖不存在!");
        }

        //获取到节点的所有的连线信息
        List<List<JobInfoDto>> allLinkedDtoList = listAllJobList(edges, ids);


        //根据连线和路由条件，得到对应的workflow,并且保存到hdfs上
        OozieClient wc = new OozieClient(oozieConfig.getUrl());
        HdfsUtil hdfsUtil = new HdfsUtil(bdpJobConfig);
        JobUtil jobUtil = new JobUtil(hdfsUtil, wc);
        Properties conf = jobUtil.createOrUpdateJobLay(procId, getQueueName(procId), workFlowDepId.intValue(), allLinkedDtoList, false, true,jobNodeConfig);

        //如果有调度，则暂停之前的调度
        if (StringUtils.isNotEmpty(workFlowDep.getLastCorOozieId())) {
            jobUtil.stopJob(workFlowDep.getLastCorOozieId());
            //改变其下的任务和节点的运行状态
            workFlowDepService.stopAllJobAndJobNodeHistoryByWorkFlowLay(workFlowDep.getId(), null, workFlowDep.getLastCorOozieId());
        }

        if (null == jobNodeConfig || null == jobNodeConfig.getSchedule_type() || jobNodeConfig.getSchedule_type() != 1) {
            if(null != hdfsUtil) {
                hdfsUtil.close();
            }
            return;
        }


        Properties autoConf = jobUtil.createOrUpdateJobLay(procId, getQueueName(procId), workFlowDepId.intValue(), allLinkedDtoList, true, true,jobNodeConfig);
        JSONObject objStart = new JSONObject();
        String oozieJobId = jobUtil.runCoordinatorJobLay(procId, workFlowDepId.intValue(), autoConf, jobNodeConfig.getExpression(), jobNodeConfig.getScheduleOrInterval().toString(),objStart,jobNodeConfig.getCronType(),jobNodeConfig.getCronUrl());

        //追加一个开始节点，让其自动回调给rest_api的start。获取到对应的任务ozzieId,方便轮询查询对应的任务的日志信息
        //jobName : type/procId/jobId      type: 0:任务  1：任务流依赖
        String beginShell = pingRestApiStartShell(oozieJobId, WorkFlowType.WORKFLOW_LAY.getCode() + "/" + procId + "/" + workFlowDepId,WorkFlowType.WORKFLOW_LAY.getCode());
        String startDst = "workflow-app/" + WorkFlowType.WORKFLOW_LAY.getCode() + "/" + procId + "/" + workFlowDepId + "/StartTwo_0.sh";
        hdfsUtil.writeFile(beginShell.getBytes(), startDst);
        if(null != hdfsUtil) {
            hdfsUtil.close();
        }

        //保存对应的任务的历史
        workFlowDep.setNum(1);
        workFlowDep.setLastCorOozieId(oozieJobId);
        String startTime = objStart.getString("startTime");
        Date date01 = new Date();
        try {
            SimpleDateFormat simpleDateFormat1 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            date01 = simpleDateFormat1.parse(startTime);
        }catch (Exception e){
            log.error("转换时间异常！error:{}",e.getMessage());
        }
        workFlowDep.setCronLastRunTime(date01);
        workFlowDepMapper.updateById(workFlowDep);
    }

    private List<List<JobInfoDto>> listAllJobList(List<JobLinkDto> edges, Set<Integer> olderNodeIds) {
        //找出所有的连线（包括虚拟的开始节点）
        List<List<Integer>> lists = listLinkedList(edges, olderNodeIds);
        if (CollectionUtil.isEmpty(lists)) {
            return null;
        }

        Set<Integer> nodeIds = new HashSet<>();
        for (List<Integer> list : lists) {
            for (int i = 1; i < list.size(); i++) {
                nodeIds.add(list.get(i));
            }
        }
        //查询到所节点的数据和对应的配置信息
        List<JobInfoDto> jobInfoDtos = jobInfoService.listAllNodeList(nodeIds);
        if (CollectionUtil.isEmpty(jobInfoDtos)) {
            return null;
        }
        //直接生成对应的表达式
        List<List<JobInfoDto>> allLinkedDtoList = new ArrayList<>();
        JobInfoDto startJobInfoDto = new JobInfoDto();
        startJobInfoDto.setWorkFlowNodeName("Start_0");
        for (List<Integer> list : lists) {
            List<JobInfoDto> list1 = new ArrayList<>();
            list1.add(startJobInfoDto);
            for (int i = 1; i < list.size(); i++) {
                int finalI = i;
                List<JobInfoDto> collect = jobInfoDtos.stream().filter(jobInfoDto -> Objects.equals(list.get(finalI), jobInfoDto.getId())).collect(Collectors.toList());
                list1.add(collect.get(0));
            }
            allLinkedDtoList.add(list1);
        }
        return allLinkedDtoList;

    }


    private void saveScheduleData(JobNodeConfig jobNodeConfig, Long workFlowMenuId, Boolean isCron) {
        if (isCron) {
            if (null == jobNodeConfig) {
                jobNodeConfig = new JobNodeConfig();
                jobNodeConfig.setSchedule_type(0);
            }
            if (null == jobNodeConfig.getSchedule_type()) {
                jobNodeConfig.setSchedule_type(0);
            }
            if (jobNodeConfig.getSchedule_type() == 1) {
                if(Objects.equals(2,jobNodeConfig.getCronType())){
                    //cron表达式
                    if(StringUtils.isEmpty(jobNodeConfig.getCronUrl())){
                        throw new RRException("crontab表达式必须填写!");
                    }
                }else{
                    jobNodeConfig.setCronType(1);
                    if (StringUtils.isEmpty(jobNodeConfig.getExpression())) {
                        throw new RRException("流程配置的调度配置的表达式不存在!");
                    }
                    if (null == jobNodeConfig.getScheduleOrInterval()) {
                        throw new RRException("流程配置的调度配置的循环方式不存在!");
                    }
                }

                //如果有配置重试，则还需要限制对应的时间
                if(Objects.equals(jobNodeConfig.getIsRetry(),1)){
                    cheackRetryData(jobNodeConfig.getReTryInterval(),jobNodeConfig.getReTryMax(),jobNodeConfig.getExpression(),jobNodeConfig.getScheduleOrInterval());
                }
            }
        }
        if (null == jobNodeConfig) {
            return;
        }
//        if (!isCron) {
//            if (CollectionUtil.isEmpty(jobNodeConfig.getSysParam())) {
//                return;
//            }
//        }
        jobNodeConfig.setJobNodeId(workFlowMenuId.intValue());
        jobNodeConfig.setJobType(1);

        //判断是否配置时间格式timeFormat，且符合规范
        if (jobNodeConfig.getTimeFormat() != null && !"".equals(jobNodeConfig.getTimeFormat())){
            TimeUtil.getDateToFormat(TimeUtil.strToDateLong(jobNodeConfig.getSysStartTime()),jobNodeConfig.getTimeFormat());
//            throw new RRException("时间格式错误，请重新配置！");
        }
        List<JobNodeConf> jobNodeConfs = JobNodeConfig.jobNodeConfigTolist(jobNodeConfig);
        jobNodeConfMapper.upsertKVByJobNodeId(jobNodeConfs);

    }

    private void cheackRetryData(Integer reTryInterval, Integer reTryMax, String expression, Integer scheduleOrInterval) {
        if(null == reTryInterval || null == reTryMax){
            throw new RRException("重试间隔时间和重试次数必须填写！");
        }
        if(reTryInterval < 0){
            throw new RRException("重试间隔时间必须大于0！");
        }
        if(reTryMax < 0 || reTryMax > 10){
            throw new RRException("重试次数必须在0~10之间！");
        }
        if(Objects.equals(ScheduleRepeatType.INTERVAL.getCode(),scheduleOrInterval)){
            JSONObject jsonObject = JSONObject.parseObject(expression);
            if(null != jsonObject){
                String type = jsonObject.getString("type");
                if(Objects.equals(type,"hour")){
                    Integer minute = jsonObject.getInteger("minute");
                    if(minute < 30){
                        throw new RRException("如果有配置重试，请设置间隔时间大于30分钟！");
                    }
                    if(minute < reTryInterval * reTryMax){
                        throw new RRException("间隔调度的时间比错误重试的间隔总时间短，请调整时间设置！");
                    }
                }
            }
        }
    }

    @Override
    public List<WorkMenuDto> listWorkFlows(Integer procId) {
        //根据类型查询到对应的结构
        List<WorkMenu> workMenus = workMenuMapper.selectList(new QueryWrapper<WorkMenu>()
                .eq("proc_id", procId)
                .in("work_type", 0, 2, 3,5));
        if (CollectionUtil.isEmpty(workMenus)) {
            return null;
        }
        //得到对应的数据
        List<WorkMenuDto> restList = null;
        try {
            restList = changeTreePackSubWork(PojoUtils.listConvert(WorkMenuDto.class, workMenus), false, false);
        } catch (IllegalAccessException e) {
            log.error("获取工作流菜单树异常，error:{}", e.getMessage());
            throw new RRException("获取工作流菜单树异常！");
        }
        return restList;
    }

    @Override
    public PageUtils getWorkFlowMenus(Map<String, Object> params) {

        List<Integer> procIdList = new ArrayList<>();
        Long userId = ShiroUtils.getUserId();
        Integer tenantId = ShiroUtils.getTenantId();
        if(sysUserRoleService.superAdmin(userId)){
            //查询到租户管理员可以看到的所有的项目
            List<ProcInfo> procInfos = procInfoMapper.listUserProcs(null, tenantId);
            if(CollectionUtil.isNotEmpty(procInfos)){
                params.put("auth",1);
                procIdList = procInfos.stream().map(ProcInfo::getId).distinct().collect(Collectors.toList());
            }else{
                params.put("auth",0);
            }
        }else{
            //查询到关联的项目
//            Map<String, Object> projectParams=new HashMap<>();
//            projectParams.put("current_per", ShiroUtils.getUserId().intValue());
//            projectParams.put("sidx", "id");
//            projectParams.put("order", "desc");
//            PageUtils pageProject = procInfoService.selectListPage(projectParams);
//            List<ProcInfo> list = (List<ProcInfo>) pageProject.getList();
//            List<Integer> procIdList= list.stream().map(e->e.getId()).distinct().collect(Collectors.toList());
            List<Integer> list = procUserMapper.procIdListByUserId(ShiroUtils.getUserId().intValue(),tenantId);
            if(CollectionUtil.isNotEmpty(list)){
                procIdList = list.stream().distinct().collect(Collectors.toList());
            }
            if (CollectionUtil.isEmpty(procIdList)){
                params.put("auth",0);

            }else{
                params.put("auth",1);

            }
        }
        IPage<WorkMenu> page = this.baseMapper.selectListPage((Page<WorkMenu>) new Query<WorkMenu>().getPage(params), params,procIdList);

        return new PageUtils(page);
    }

    @Override
    public Integer copyVirJobNode(Integer jobNodeId) {
        JobNodeInfo jobNodeInfo = jobNodeInfoMapper.selectById(jobNodeId);
        if (null == jobNodeInfo) {
            throw new RRException("待复制的节点任务不存在!");
        }
        JobInfo jobInfo = jobInfoService.getById(jobNodeInfo.getJobId());
        if(jobInfo == null){
            throw new RRException("待复制的节点任务对应的工作流不存在!");
        }
        //验证对应的用户是否可以操作这个工程
        cheackProcId(jobInfo.getProcId(), ShiroUtils.getUserId(), false, null,null,null);

        WorkMenuJobRel workMenuJobRel = workMenuJobRelMapper.
                selectOne(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 1).eq("work_flow_id", jobInfo.getId()));
        if (null == workMenuJobRel) {
            throw new RRException("任务和默认作业流菜单关联不存在!");
        }

        JobNodeInfo newJobNodeInfo = new JobNodeInfo();
        BeanUtils.copyProperties(jobNodeInfo,newJobNodeInfo);
        newJobNodeInfo.setId(null);
        newJobNodeInfo.setNodeKey(null);
        newJobNodeInfo.setPNodeKey(null);
        newJobNodeInfo.setPosY(null);
        newJobNodeInfo.setPosX(null);
        newJobNodeInfo.setEnable(JobStatus.ACTIVE.getCode());
        newJobNodeInfo.setCreateTime(DateTime.now());
        newJobNodeInfo.setCreatePer(ShiroUtils.getUserId().intValue());
        String newName = jobNodeId+ "&&&&&&" + UUID.randomUUID();
        newJobNodeInfo.setName(newName);
        newJobNodeInfo.setIsVirCopy(1);
        newJobNodeInfo.setIsVirCopyExpire(System.currentTimeMillis());
        jobNodeInfoMapper.insert(newJobNodeInfo);

        //添加节点关联表数据
        WorkMenuJobRel workMenuJobRel1 = new WorkMenuJobRel();
        BeanUtils.copyProperties(workMenuJobRel, workMenuJobRel1);
        workMenuJobRel1.setId(null);
        workMenuJobRel1.setWorkFlowId(Long.valueOf(newJobNodeInfo.getId()));
        workMenuJobRel1.setWorkType(0);
        workMenuJobRelMapper.insert(workMenuJobRel1);

        //服务器地址的复制
        if(Objects.equals(jobNodeInfo.getTypeId(),JobType.SHELL.getCode())){
            BussessConfigEntity bussessConfigEntity = bussessConfigMapper.selectOne(new QueryWrapper<BussessConfigEntity>()
                    .eq("bussess_type", 1).eq("bussess_id", jobNodeInfo.getId()));
            if(null != bussessConfigEntity){
                bussessConfigEntity.setId(null);
                bussessConfigEntity.setBussessId(newJobNodeInfo.getId());
                bussessConfigMapper.insert(bussessConfigEntity);
            }
        }

        //复制需要复制的数据集合
        //还得复制对应的配置的sql语句和对应的配置信息
        copyJobNodeConfig(newJobNodeInfo.getId(), jobNodeInfo.getId());

        return newJobNodeInfo.getId();
    }

    @Override
    public void deleteVirJobNode(Integer jobNodeId) {
        JobNodeInfo jobNodeInfo = jobNodeInfoMapper.selectById(jobNodeId);
        if(null != jobNodeInfo){
            if(Objects.equals(1,jobNodeInfo.getIsVirCopy())){
                deleteVirNode(jobNodeId);
            }
        }
    }

    private void deleteVirNode(Integer jobNodeId) {
        workMenuJobRelMapper.delete(new QueryWrapper<WorkMenuJobRel>().eq("work_type",0).eq("work_flow_id",jobNodeId));
        bussessConfigMapper.delete(new QueryWrapper<BussessConfigEntity>()
                .eq("bussess_type", 1).eq("bussess_id", jobNodeId));
        jobNodeInfoMapper.deleteById(jobNodeId);
        jobNodeConfMapper.delete(new QueryWrapper<JobNodeConf>().eq("job_node_id",jobNodeId).eq("job_type",0));
    }

    @Override
    public void moveVirJobNode() {
        List<JobNodeInfo> jobNodeInfos = jobNodeInfoMapper.selectList(new QueryWrapper<JobNodeInfo>()
                .eq("is_vir_copy",1).isNotNull("is_vir_copy_expire"));
        if(CollectionUtil.isNotEmpty(jobNodeInfos)){
            long currents = System.currentTimeMillis();
            List<JobNodeInfo> jobs = jobNodeInfos.stream().filter(o -> (currents - o.getIsVirCopyExpire() > 12 * 60 * 1000)).collect(Collectors.toList());
            if(CollectionUtil.isNotEmpty(jobs)){
                for (JobNodeInfo job : jobs) {
                    deleteVirNode(job.getId());
                }
            }
        }
    }

    @Override
    public List<ProcInfo> procs(Long userId, Integer tenantId) {
        if(sysUserRoleService.superAdmin(userId)){
            return workMenuMapper.procs(userId,null);
        }else{
            return workMenuMapper.procs(userId,tenantId);
        }
    }

    @Override
    public Integer getProIdByMenuId(Long jobMenuId) {
        WorkMenu workMenu = workMenuMapper.selectById(jobMenuId);
        if(workMenu != null){
            return workMenu.getProcId();
        }
        return null;
    }

    @Override
    public List<WorkMenuDto> getProjectWorkFlowMenues() {

        //获取项目列表
        Map<String,Object> params = new HashMap<>();
        Long userId = ShiroUtils.getUserId();
        Integer tenantId = ShiroUtils.getTenantId();
        if(sysRoleService.isSuperAdmin(userId)){
            params.put("userId", null);
        }else{
            params.put("userId", userId);
        }
        params.put("tenantId", tenantId);//租户id
        params.put("currentUserId",userId);
        //如果不是租户管理员，值查询关联这个项目的用户
        List<ProcInfoListRes> list = procInfoMapper.selectProcOwer(params);

        //遍历项目，获取任务树
        List<WorkMenuDto> dataList = new ArrayList<>();
        if (list != null && list.size() != 0) {

            for (ProcInfoListRes res : list) {
                if (res.getAuth()) {
                    WorkMenuDto menuDto = new WorkMenuDto();
                    menuDto.setName(res.getName());
                    menuDto.setId(Long.valueOf(res.getId()));
                    menuDto.setChildren(listWorkFlowMenues(res.getId()));
                    dataList.add(menuDto);
                }
            }
        }
        return dataList;
    }

    @Override
    public void copyWorkJob(WorkMenuDto workMenuDto) {

        //验证参数
        if (null == workMenuDto) {
            throw new RRException(DataDevelopmentBizExceptionEnum.PARAM_ERROR.getMessage());
        }
        if (null == workMenuDto.getId()) {
            throw new RRException("待复制的节点任务id必须传递!");
        }
        WorkMenu workMenu = workMenuMapper.selectById(workMenuDto.getWorkFlowId());
        if (null == workMenu) {
            throw new RRException("默认作业流不存在!");
        }
        if (WorkFlowMenuType.WORKFLOW_FIELD.getCode() != workMenu.getWorkType()) {
            throw new RRException("上级目录不是默认作业流类型!");
        }
        //验证是否可以添加默认作业流
        if (null == workMenuDto.getProcId()) {
            throw new RRException("工程id不能为空!");
        }
        //验证对应的用户是否可以操作这个工程
        cheackProcId(workMenuDto.getProcId(), ShiroUtils.getUserId(), false, null,null,null);
        //名字必须传
        if (StringUtils.isEmpty(workMenuDto.getName())) {
            throw new RRException("目录名字必须传递!");
        }
        if (null == workMenuDto.getTypeId()) {
            throw new RRException("任务类型必须传递!");
        }
        JobNodeInfo jobNodeInfo = jobNodeInfoMapper.selectById(workMenuDto.getId());
        if (null == jobNodeInfo) {
            throw new RRException("待复制的节点任务不存在!");
        }
        if (!workMenuDto.getTypeId().equals(jobNodeInfo.getTypeId())) {
            throw new RRException("待复制的节点任务和新增的任务节点类型不一致!");
        }

        //验证节点的名字是否重复
        WorkMenuJobRel workMenuJobRel = workMenuJobRelMapper.selectOne(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 1).eq("work_menu_id", workMenuDto.getWorkFlowId()));
        if (null == workMenuJobRel) {
            throw new RRException("任务和默认作业流菜单关联不存在!");
        }
        JobNodeInfo jobNodeInfo1 = workMenuJobRelMapper.selectJobNodeInfoByMenuIdAndName(workMenuDto.getWorkFlowId().intValue(),workMenuDto.getName(),workMenuDto.getProcId());
        if (null != jobNodeInfo1) {
            throw new RRException("同一作业流菜单下的任务名称不能有重复!");
        }

        JobNodeInfo newJobNodeInfo = new JobNodeInfo();
        newJobNodeInfo.setName(workMenuDto.getName());
        newJobNodeInfo.setTypeId(workMenuDto.getTypeId());
        newJobNodeInfo.setEnable(JobStatus.ACTIVE.getCode());
        newJobNodeInfo.setCreateTime(DateTime.now());
        newJobNodeInfo.setCreatePer(ShiroUtils.getUserId().intValue());
        newJobNodeInfo.setJobId(workMenuJobRel.getWorkFlowId().intValue());
        newJobNodeInfo.setNodeKey(null);
        newJobNodeInfo.setPNodeKey(null);
        newJobNodeInfo.setDesc(jobNodeInfo.getDesc());
        jobNodeInfoMapper.insert(newJobNodeInfo);

        //添加节点关联表数据
        WorkMenuJobRel workMenuJobRel1 = new WorkMenuJobRel();
        BeanUtils.copyProperties(workMenuJobRel, workMenuJobRel1);
        workMenuJobRel1.setId(null);
        workMenuJobRel1.setWorkFlowId(Long.valueOf(newJobNodeInfo.getId()));
        workMenuJobRel1.setWorkType(0);
        workMenuJobRelMapper.insert(workMenuJobRel1);

        //服务器地址的复制
        if(Objects.equals(workMenuDto.getTypeId(),JobType.SHELL.getCode())){
            BussessConfigEntity bussessConfigEntity = bussessConfigMapper.selectOne(new QueryWrapper<BussessConfigEntity>()
                    .eq("bussess_type", 1).eq("bussess_id", jobNodeInfo.getId()));
            if(null != bussessConfigEntity){
                bussessConfigEntity.setId(null);
                bussessConfigEntity.setBussessId(newJobNodeInfo.getId());
                bussessConfigMapper.insert(bussessConfigEntity);
            }
        }

        //还得复制对应的配置的sql语句和对应的配置信息
        copyJobNodeConfig(newJobNodeInfo.getId(), jobNodeInfo.getId());

        // 虚拟化一个开始节点和一个结束节点，并且保存到hdfs上
        saveStartNodeAndEndNode(workMenuDto.getProcId().longValue(), workMenuJobRel.getWorkFlowId().intValue(), WorkFlowType.WORKFLOW.getCode());


    }

    @Override
    public List<JobNodeConf> listWorkFlowConfig(Long id) {
        return jobNodeConfMapper.selectList(new QueryWrapper<JobNodeConf>()
                .eq("job_type", 1)
                .eq("job_node_id", id)
                .in("`key`",
                        "schedule_type","scheduleOrInterval","expression",
                        "sysParam",
                        "nodeParam",
                        "isDurTime","startTime","endTime",
                        "isRetry","reTryMax","reTryInterval",
                        "isOpenTime","sysStartTime","sysEndTime","sysDurTime","cronType","cronUrl","timeFormat"));
    }


    private void copyJobNodeConfig(Integer newJobNodeId, Integer olderJobNodeId) {
//        List<JobNodeConf> rest = new ArrayList<>();
//        //SQL计算
//        List<JobNodeConf> jobNodeConfs = jobNodeConfMapper.selectList(new QueryWrapper<JobNodeConf>()
//                .eq("job_node_id", olderJobNodeId)
//                .eq("job_type",0)
//        .eq("`key`","sql_statment"));
//        if(CollectionUtil.isNotEmpty(jobNodeConfs)){
//            rest.addAll(jobNodeConfs);
//        }
//        //输入设置_输入内容
//        List<JobNodeConf> jobNodeConfs1 = jobNodeConfMapper.selectList(new QueryWrapper<JobNodeConf>()
//                .eq("job_node_id", olderJobNodeId)
//                .eq("job_type",0)
//                .eq("`key`","input_input_content"));
//        if(CollectionUtil.isNotEmpty(jobNodeConfs1)){
//            rest.addAll(jobNodeConfs1);
//        }
//        //输出设置_前置语句
//        List<JobNodeConf> jobNodeConfs2 = jobNodeConfMapper.selectList(new QueryWrapper<JobNodeConf>()
//                .eq("job_node_id", olderJobNodeId)
//                .eq("job_type",0)
//                .eq("`key`","output_pre_statment"));
//        if(CollectionUtil.isNotEmpty(jobNodeConfs2)){
//            rest.addAll(jobNodeConfs2);
//        }
//        //输出设置_前置语句
//        List<JobNodeConf> jobNodeConfs3 = jobNodeConfMapper.selectList(new QueryWrapper<JobNodeConf>()
//                .eq("job_node_id", olderJobNodeId)
//                .eq("job_type",0)
//                .eq("`key`","output_pre_statment"));
//        if(CollectionUtil.isNotEmpty(jobNodeConfs3)){
//            rest.addAll(jobNodeConfs3);
//        }
        List<JobNodeConf> jobNodeConfs = new ArrayList<>();
        List<JobNodeConf> jobNodeConfs1 = jobNodeConfMapper.selectList(new QueryWrapper<JobNodeConf>()
                .eq("job_node_id", olderJobNodeId)
                .eq("job_type",0)
                .ne("`key`","priorityConfig")
                .ne("`key`","jobOutPutTableIds")
                .ne("`key`","jobNodeId"));
        if(CollectionUtil.isNotEmpty(jobNodeConfs1)){
            jobNodeConfs.addAll(jobNodeConfs1);
        }
        JobNodeConf jobNodeConf=new JobNodeConf();
        jobNodeConf.setJobType(0);
        jobNodeConf.setJobNodeId(newJobNodeId);
        jobNodeConf.setKey("jobNodeId");
        jobNodeConf.setValue(newJobNodeId.toString());
        jobNodeConfs.add(jobNodeConf);
        if (CollectionUtil.isNotEmpty(jobNodeConfs)) {
            for (JobNodeConf j : jobNodeConfs) {
                j.setId(null);
                j.setJobNodeId(newJobNodeId);
            }
            jobNodeConfMapper.upsertKVByJobNodeId(jobNodeConfs);
        }

    }


    private void delMenu(WorkMenu workMenu, Map<String, Set<Long>> allChildrens) throws Exception {
        //如果是作业流依赖菜单则删除所有相关作业流依赖日志以及相关作业流日志
        if (workMenu.getWorkType().equals(WorkFlowMenuType.WORKFLOW_MGT_FIELD.getCode())){
            WorkFlowDep workFlowDep = workFlowDepMapper.selectOne(new QueryWrapper<WorkFlowDep>().eq("work_menu_id", workMenu.getId()));
            if (workFlowDep!=null){
                List<String> depOozieIds=jobDepRunHistoryMapper.selectOozieIdByDepId(workFlowDep.getId());
                if(CollectionUtil.isNotEmpty(depOozieIds)){
                    jobRunHistoryService.remove(new QueryWrapper<JobRunHistory>().in("dep_oozie_job_id",depOozieIds));
                    List<String> oozieIds =jobRunHistoryService.selectOozieIdsByDepOozieIds(depOozieIds);
                    if(CollectionUtil.isNotEmpty(oozieIds)){
                        jobNodeRunHistoryService.remove(new QueryWrapper<JobNodeRunHistory>().in("oozie_job_id",oozieIds));
                    }
                }
                jobDepRunHistoryMapper.delete(new QueryWrapper<JobDepHistory>().eq("job_dep_id",workFlowDep.getId()));
                //根据job_node_id删除节点配置表
                jobNodeConfMapper.delete(new QueryWrapper<JobNodeConf>().eq("job_node_id", workMenu.getId()).eq("job_type",1));
                workFlowDepMapper.deleteById(workFlowDep.getId());
            }
        }
        //如果是作业流菜单则删除所有相关作业流日志
        else if(workMenu.getWorkType().equals(WorkFlowMenuType.WORKFLOW_FIELD.getCode())){
            WorkMenuJobRel workMenuJobRel = workMenuJobRelMapper.selectOne(new QueryWrapper<WorkMenuJobRel>().eq("work_type", WorkFlowType.WORKFLOW.getCode()).eq("work_menu_id", workMenu.getId()));
            if (workMenuJobRel!=null){
                jobInfoService.deleteAllOfJob(workMenuJobRel.getWorkFlowId().intValue(),ShiroUtils.getUserId());
                workMenuJobRelMapper.deleteById(workMenuJobRel.getId());
//                jobRunHistoryService.remove(new QueryWrapper<JobRunHistory>().eq("job_info_id",workMenuJobRel.getWorkFlowId()));
//                jobNodeRunHistoryService.remove(new QueryWrapper<JobNodeRunHistory>().eq("job_info_id",workMenuJobRel.getWorkFlowId()));
//                jobInfoService.removeById(workMenuJobRel.getWorkFlowId());

            }

        }

        if (0L == workMenu.getParentId()) {
            if (null != allChildrens) {
                Set<Long> menuSet = allChildrens.get("menuSet");
                if (CollectionUtil.isNotEmpty(menuSet)) {
                    //查询出来对应的工作流菜单
                    List<WorkMenuJobRel> lists = workMenuJobRelMapper.listWFJobs(menuSet);
                    if(CollectionUtil.isNotEmpty(lists)){
                        List<Long> jobIds = lists.stream().map(WorkMenuJobRel::getWorkFlowId).collect(Collectors.toList());
                        jobInfoMapper.deleteBatchIds(jobIds);
                        List<Long> ids = lists.stream().map(WorkMenuJobRel::getId).collect(Collectors.toList());
                        workMenuJobRelMapper.deleteBatchIds(ids);
                    }
                    workMenuMapper.deleteBatchIds(menuSet);
                }
            }
        }
        //删除对应的工作流依赖关联表
        if (CollectionUtil.isNotEmpty(allChildrens.get("jobSet"))) {
            //查询到这个
            Set<Long> menuSets = new HashSet<>();
            for (Long aLong : allChildrens.get("jobSet")) {
                WorkMenuJobRel workMenuJobRel = workMenuJobRelMapper.selectOne(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 2).eq("work_flow_id", aLong));
                if (null != workMenuJobRel) {
                    workMenuJobRelMapper.deleteById(workMenuJobRel.getId());
                    menuSets.add(workMenuJobRel.getWorkMenuId());
                }
            }
            if (CollectionUtil.isNotEmpty(menuSets)) {
                List<WorkFlowDep> workFlowDeps = workFlowDepMapper.selectByworkFlowDepList(menuSets);
                if (CollectionUtil.isNotEmpty(workFlowDeps)) {
                    workFlowDepMapper.deleteById(workFlowDeps.get(0).getId());
                }
            }
        }
        workMenuMapper.deleteById(workMenu.getId());

        WorkFlowDep oldDep = workFlowDepMapper.selectOne(new QueryWrapper<WorkFlowDep>().eq("work_menu_id", workMenu.getId()));
        if(oldDep != null)
            workFlowDepMapper.deleteById(oldDep.getId());
    }

    private Map<String, Set<Long>> cheackMenu(WorkMenu workMenu) {
        Map<String, Set<Long>> allChildrens = null;
        //如果是工作流菜单目录
        if(Objects.equals(WorkFlowMenuType.WORKFLOW_FIELD.getCode(),workMenu.getWorkType())){
            List<WorkMenuJobRel> workMenuJobRels = workMenuJobRelMapper.selectList(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 0).eq("work_menu_id", workMenu.getId()));
            if(CollectionUtil.isNotEmpty(workMenuJobRels)){
                for (WorkMenuJobRel workMenuJobRel : workMenuJobRels) {
                    JobNodeInfo jobNodeInfo = jobNodeInfoMapper.selectById(workMenuJobRel.getWorkFlowId());
                    if(null != jobNodeInfo && null != jobNodeInfo.getPosX()){
                        throw new RRException(workMenu.getName()+"工作流已配置节点,不能删除！");
                    }
                }
            }
            WorkMenuJobRel workMenuJobRel2 = workMenuJobRelMapper.selectOne(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 1).eq("work_menu_id", workMenu.getId()));
            List<WorkMenuJobRel> workMenuJobRels1 = workMenuJobRelMapper.selectList(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 2).eq("work_flow_id", workMenuJobRel2.getWorkFlowId()));
            if (CollectionUtil.isNotEmpty(workMenuJobRels1)){
                throw new RRException(workMenu.getName()+"工作流已配置工作流依赖,不能删除！");
            }

        }

        if(Objects.equals(WorkFlowMenuType.WORKFLOW_MGT_FIELD.getCode(),workMenu.getWorkType())){
            List<WorkMenuJobRel> workMenuJobRels = workMenuJobRelMapper.selectList(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 2).eq("work_menu_id", workMenu.getId()));
            if(CollectionUtil.isNotEmpty(workMenuJobRels)){
                throw new RRException(workMenu.getName()+"工作流依赖已配置工作流,不能删除！");
            }
        }


        try {
            allChildrens = listAllChildren(workMenu, workMenu.getProcId());
        } catch (IllegalAccessException e) {
            log.error("获取工作流菜单旗下所有的子节点异常，error:{}", e.getMessage());
            throw new RRException("获取工作流菜单旗下所有的子节点异常！");
        }

        if (null != allChildrens) {
            //验证任务是否已配置在工作流依赖中
            cheackJobIsHave(allChildrens.get("jobSet"));
            //验证节点是否已配置在工作流中
            cheackJobNodeIsHave(allChildrens.get("jobNodeSet"));
        }

        if (0L == workMenu.getParentId()) {
            //查看根目录下是否有节点和血缘关系
            if (null != allChildrens
                    && (CollectionUtil.isNotEmpty(allChildrens.get("jobNodeSet"))
            )) {
                throw new RRException("根目录菜单下存在任务或者工作流，请先删除任务或者作业流！");
            }
            List<WorkMenu> workMenus = workMenuMapper.selectList(new QueryWrapper<WorkMenu>()
                    .eq("root_id",workMenu.getId())
                    .eq("work_type",WorkFlowMenuType.WORKFLOW_FIELD.getCode())
                    .or()
                    .eq("root_id",workMenu.getId())
                    .eq("work_type",WorkFlowMenuType.WORKFLOW_MGT_FIELD.getCode()));
            if (CollectionUtil.isNotEmpty(workMenus)){
                for (WorkMenu w:workMenus){
                    try {
                        cheackMenu(w);
                    }catch (Exception e){
                        throw new RRException("根目录菜单下存在任务或者工作流，请先删除任务或者作业流！");
                    }
                }
            }
        } else {
            if (null != allChildrens
                    && (CollectionUtil.isNotEmpty(allChildrens.get("jobNodeSet"))
                    || CollectionUtil.isNotEmpty(allChildrens.get("menuSet")))) {
                cheackJobIsHave(allChildrens.get("jobSet"));
                throw new RRException("当前目录菜单下存在任务或者工作流或者子目录，请先删除任务或者作业流或者子目录！");
            }
        }
        return allChildrens;
    }

    private void cheackJobNodeIsHave(Set<Long> jobNodeSet) {
        if(CollectionUtil.isNotEmpty(jobNodeSet)){
            Set<Integer> jobNodes = new HashSet<>();
            for (Long aLong : jobNodeSet) {
                jobNodes.add(aLong.intValue());
            }
            List<JobNodeInfoDto> jobNodeInfoDtos = jobNodeInfoMapper.selectListByIds(jobNodes);
            if(CollectionUtil.isNotEmpty(jobNodeInfoDtos)){
                throw new RRException("工作流节点:"+jobNodeInfoDtos.get(0).getName()+"存在于工作流中不能删除！");
            }
        }
    }

    private void cheackJobIsHave(Set<Long> jobSet) {
        if (CollectionUtil.isNotEmpty(jobSet)) {
            //查询到对应的数据是否满足条件
            Set<Long> menuSets = new HashSet<>();
            for (Long aLong : jobSet) {
                WorkMenuJobRel workMenuJobRel = workMenuJobRelMapper.selectOne(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 1).eq("work_flow_id", aLong));
                if (null != workMenuJobRel) {
                    menuSets.add(workMenuJobRel.getWorkMenuId());
                }
            }
            if (CollectionUtil.isNotEmpty(menuSets)) {
                List<WorkFlowDep> workFlowDeps = workFlowDepMapper.selectByworkFlowDepList(menuSets);
                if (CollectionUtil.isNotEmpty(workFlowDeps)) {
                    //不管这个工作流是否运行，只要出现子画布中，则不能删除
                    throw new RRException("工作流依赖id:"+workFlowDeps.get(0).getId()+"中存在工作流不能删除！");
                }
            }
        }
    }

    private Map<String, Set<Long>> listAllChildren(WorkMenu workMenu, Integer procId) throws IllegalAccessException {
        //根据类型查询到对应的结构
        List<WorkMenu> workMenus = workMenuMapper.selectList(new QueryWrapper<WorkMenu>()
                .eq("proc_id", procId));
        return changeTreePackData(PojoUtils.listConvert(WorkMenuDto.class, workMenus), workMenu.getId());
    }

    private WorkMenu saveMenu(WorkMenuDto workMenuDto) {
        //添加的目录的类型
        if (Objects.equals(0L, workMenuDto.getParentId())) {
            WorkMenu workMenu = new WorkMenu();
            BeanUtils.copyProperties(workMenuDto, workMenu);
            //如果是根目录的话，那么就默认是工作流程目录类型
            workMenu.setWorkType(WorkFlowMenuType.ROOT_DIR.getCode());
            workMenu.setCreatePer(null == workMenuDto.getCreatePer() ? ShiroUtils.getUserId().intValue() : workMenuDto.getCreatePer());
            workMenuMapper.insert(workMenu);
            workMenu.setRootId(workMenu.getId());
            workMenuMapper.updateById(workMenu);

            //默认初始化对应的数据
            WorkMenu workMenu1 = new WorkMenu();
            workMenu1.setName("数据开发");
            workMenu1.setProcId(workMenuDto.getProcId());
            workMenu1.setWorkType(WorkFlowMenuType.SUB_DIR.getCode());
            workMenu1.setParentId(workMenu.getId());
            workMenu1.setRootId(workMenu.getId());
            workMenu1.setCreatePer(null == workMenuDto.getCreatePer() ? ShiroUtils.getUserId().intValue() : workMenuDto.getCreatePer());
            workMenuMapper.insert(workMenu1);

            WorkMenu workMenu2 = new WorkMenu();
            workMenu2.setProcId(workMenuDto.getProcId());
            workMenu2.setName("作业流依赖管理");
            workMenu2.setWorkType(WorkFlowMenuType.WORKFLOW_MGT.getCode());
            workMenu2.setParentId(workMenu.getId());
            workMenu2.setRootId(workMenu.getId());
            workMenu2.setCreatePer(null == workMenuDto.getCreatePer() ? ShiroUtils.getUserId().intValue() : workMenuDto.getCreatePer());
            workMenuMapper.insert(workMenu2);
            return workMenu2;

        } else {

            WorkMenu workMenu = new WorkMenu();
            BeanUtils.copyProperties(workMenuDto, workMenu);
            WorkMenu olderWorkMenu = workMenuMapper.selectById(workMenuDto.getParentId());
            if (null == olderWorkMenu) {
                throw new RRException("父级工作流不存在!");
            }
            if (Objects.equals(olderWorkMenu.getWorkType(), WorkFlowMenuType.ROOT_DIR.getCode())) {
                workMenu.setWorkType(WorkFlowMenuType.SUB_DIR.getCode());
            }
            if (Objects.equals(olderWorkMenu.getWorkType(), WorkFlowMenuType.WORKFLOW_MGT.getCode())) {
                workMenu.setWorkType(WorkFlowMenuType.WORKFLOW_MGT_FIELD.getCode());
            }
            if (Objects.equals(olderWorkMenu.getWorkType(), WorkFlowMenuType.SUB_DIR.getCode())
            || Objects.equals(olderWorkMenu.getWorkType(), WorkFlowMenuType.SUB_SUB_DIR.getCode())) {
                workMenu.setWorkType(WorkFlowMenuType.SUB_SUB_DIR.getCode());
            }
            workMenu.setCreatePer(null == workMenuDto.getCreatePer() ? ShiroUtils.getUserId().intValue() : workMenuDto.getCreatePer());
            workMenuMapper.insert(workMenu);

            //新增的作业流血缘依赖文件
            if (Objects.equals(olderWorkMenu.getWorkType(), WorkFlowMenuType.WORKFLOW_MGT.getCode())) {
                WorkFlowDep workFlowDep = new WorkFlowDep();
                workFlowDep.setProcId(workMenuDto.getProcId());
                workFlowDep.setParam(null);
                workFlowDep.setWorkMenuId(workMenu.getId());
                workFlowDep.setLastRunState(null);
                workFlowDep.setCreateTime(LocalDateTime.now());
                workFlowDep.setDescName(workMenuDto.getDescName());
                workFlowDep.setJobLayStatus(0);
                workFlowDep.setCreatePer(null == workMenuDto.getCreatePer() ? ShiroUtils.getUserId().intValue() : workMenuDto.getCreatePer());
                workFlowDepMapper.insert(workFlowDep);

                //生成对应的虚拟化开始节点和结束节点的shell
                saveStartJobNodeAndEndJobNode(workMenuDto.getProcId().longValue(), workFlowDep.getId().intValue(), WorkFlowType.WORKFLOW_LAY.getCode());
            }

            return workMenu;
        }

    }

    /***
     * 验证当前工程，当前用户是否可以操作
     * @param procId  工程id
     * @param userId  用户id
     * @param isMangerment  是否是创建者才能操作
     * @param createPer  创建的用户  如果为null  则判断其它的用户  如果不为null 必须判断当前用户和创建者一样
     * @param userRole  不需要核查权限的角色 为1时，超级管理员  如果为null  则判断其它的用户  如果不为null 必须判断当前用户和创建者一样
     */
    @Override
    public void cheackProcId(Integer procId, Long userId, Boolean isMangerment, Integer createPer,String userRole,Integer isImport) {
        //查询到对应的工程信息
        ProcInfo procInfo = procInfoMapper.selectById(procId);
        if (null == procInfo) {
            throw new RRException("工程不存在!");
        }
        if (isMangerment) {
            if (userRole.equals("admin")) {
                if (sysUserRoleService.superAdmin(userId)) {
                    return;
                }
                if(Objects.equals(isImport,1)){
                    //如果是导入，则当前操作者如果是项目管理员也可以导入成功
                    ProcUser procUser = procUserMapper.selectOne(new QueryWrapper<ProcUser>().eq("proc_id", procId).eq("is_owner",1));
                    if(null != procUser){
                        if(userId.intValue() == procUser.getUserId()){
                            return;
                        }
                    }
                }
                if (createPer != userId.intValue()) {
                    throw new RRException("当前用户无权操作!");
                }
            } else if (userRole.equals("user")) {
                if (sysUserRoleService.superAdmin(userId) || sysUserRoleService.admin(userId) || Objects.equals(procInfo.getCreatePer(), userId.intValue())) {
                    return;
                }
                if (createPer != userId.intValue()) {
                    throw new RRException("当前用户无权操作!");
                }
            }
        } else {
            //是否是这个项目的成员
            List<ProcUser> procUsers = procUserMapper.selectList(new QueryWrapper<ProcUser>().eq("proc_id", procId));
            if (CollectionUtil.isEmpty(procUsers)) {
                throw new RRException("当前用户无权操作这个工程!");
            }
            Set<Integer> userIds = new HashSet<>();
            procUsers.forEach(procUser -> {
                userIds.add(procUser.getUserId());
            });
            if (!userIds.contains(userId.intValue())) {
                throw new RRException("当前用户无权操作这个工程!");
            }
        }
    }

    @Override
    public String getJobLayOozieId(Long workMenuDepId) {
        //查询到对应的手动的oozieId
        WorkFlowDep workFlowDep = workFlowDepMapper.selectById(workMenuDepId);
        if(null == workFlowDep){
            return "-1";
        }
        return StringUtils.isEmpty(workFlowDep.getLastOozieId()) ? "-1" : workFlowDep.getLastOozieId();
    }

    @Override
    public List<Integer> initJobLayStartShell() {
        //查询到都有的工作流依赖菜单
        List<WorkFlowDep> workFlowDeps = workFlowDepMapper.selectList(new QueryWrapper<WorkFlowDep>());
        List<Integer> ids = new ArrayList<>();
        if(CollectionUtil.isNotEmpty(workFlowDeps)){
            for (WorkFlowDep workFlowDep : workFlowDeps) {
                try {
                    String nodeName = JobType.ObjOf(0).getEnName();
                    String dst = "workflow-app/" + WorkFlowType.WORKFLOW_LAY.getCode() + "/" + workFlowDep.getProcId() + "/" + workFlowDep.getId() + "/" + nodeName + "_0.sh";
                    jobNodeInfoService.uploadJobShell(0, dst, oozieConfig,workFlowDep.getId().intValue());
                }catch (Exception e){
                    ids.add(workFlowDep.getId().intValue());
                }
            }
        }
        return ids;
    }

    @Override
    public Map<String,List<Integer>> initJobLayStartAutoShell() {
        //查询到都有的工作流依赖菜单
        List<WorkFlowDep> workFlowDeps = workFlowDepMapper.selectList(new QueryWrapper<WorkFlowDep>());
        Map<String,List<Integer>> restMap = new HashMap<>();
        //自动的开始shell重新生成异常
        List<Integer> changeIds = new ArrayList<>();
        List<Integer> closeSuccessIds = new ArrayList<>();
        List<Integer> closeErrorIds = new ArrayList<>();
        List<Integer> saveUpdateIds = new ArrayList<>();
        if(CollectionUtil.isNotEmpty(workFlowDeps)){
            for (WorkFlowDep workFlowDep : workFlowDeps) {
                //查询对应的调度方式和系统参数
                List<JobNodeConf> jobNodeConfs = jobNodeConfMapper.selectList(new QueryWrapper<JobNodeConf>()
                        .eq("job_node_id", workFlowDep.getWorkMenuId()).eq("job_type", 1));
                JobNodeConfig jobNodeConfig = null;
                if (CollectionUtil.isNotEmpty(jobNodeConfs)) {
                    JobNodeConf olderJobNodeConf = null;
                    List<JobNodeConf> newJobNodeConfs = new ArrayList<>();
                    for (JobNodeConf jobNodeConf : jobNodeConfs) {
                        if ("sysParam".equals(jobNodeConf.getKey())) {
                            olderJobNodeConf = jobNodeConf;
                        } else {
                            newJobNodeConfs.add(jobNodeConf);
                        }
                    }
                    if (CollectionUtil.isNotEmpty(newJobNodeConfs)) {
                        jobNodeConfig = JobNodeConfig.listToJobNodeConfig(newJobNodeConfs);
                    }
                    //系统参数
                    if (null != olderJobNodeConf) {
                        if (null == jobNodeConfig) {
                            jobNodeConfig = new JobNodeConfig();
                        }
                        JSONArray jsonArray = JSONArray.parseArray(olderJobNodeConf.getValue());
                        if (CollectionUtil.isNotEmpty(jsonArray)) {
                            List<Map<String, Object>> restList = new ArrayList<>();
                            for (Object obj : jsonArray) {
                                Map<String, Object> values = JSONObject.parseObject(obj.toString()).getInnerMap();
                                restList.add(values);
                            }
                            jobNodeConfig.setSysParam(restList);
                        }
                    }
                }
                if (null != jobNodeConfig && jobNodeConfig.getSchedule_type() == 1) {
                    //直接编辑保存这个作业流依赖
                    jobNodeConfig.setSchedule_type(0);
                    changeIds.add(workFlowDep.getId().intValue());
                }
                //禁用作业流
                try {
                    if(Objects.equals(1,workFlowDep.getJobLayStatus())){
                        closeWorkFlowJob(workFlowDep.getWorkMenuId());
                        closeSuccessIds.add(workFlowDep.getId().intValue());
                    }
                }catch (Exception e){
                    closeErrorIds.add(workFlowDep.getId().intValue());
                }
                //重新编辑保存对应的任务
                try {
                    JobLinkAndJobDto jobLinkAndJobDto = packSaveOrUpdateData(jobNodeConfig,workFlowDep);
                    if(null != jobLinkAndJobDto){
                        updateWorkFlow(jobLinkAndJobDto, Boolean.TRUE);
                    }
                }catch (Exception e){
                    saveUpdateIds.add(workFlowDep.getId().intValue());
                }
//                try {
//                    HdfsUtil hdfsUtil = new HdfsUtil(bdpJobConfig);
//                    //追加一个开始节点，让其自动回调给rest_api的start。获取到对应的任务ozzieId,方便轮询查询对应的任务的日志信息
//                    //jobName : type/procId/jobId      type: 0:任务  1：任务流依赖
//                    String beginShell = pingRestApiStartShell(workFlowDep.getLastCorOozieId(), WorkFlowType.WORKFLOW_LAY.getCode() + "/" + workFlowDep.getProcId() + "/" + workFlowDep.getId(),WorkFlowType.WORKFLOW_LAY.getCode());
//                    String startDst = "workflow-app/" + WorkFlowType.WORKFLOW_LAY.getCode() + "/" + workFlowDep.getProcId() + "/" + workFlowDep.getId() + "/StartTwo_0.sh";
//                    hdfsUtil.writeFile(beginShell.getBytes(), startDst);
//                }catch (Exception e){
//                    ids.add(workFlowDep.getId().intValue());
//                }
            }
            restMap.put("定时调度改为暂停调度作业流依赖", changeIds);
            restMap.put("禁用作业流依赖成功", closeSuccessIds);
            restMap.put("禁用作业流依赖失败", closeErrorIds);
            restMap.put("编辑保存作业流依赖异常", saveUpdateIds);
        }
        return restMap;
    }

    @Override
    public List<ProcWithWorkMenusVO> procWithWorkMenus(Integer type) {
        List<ProcWithWorkMenusVO> resultList = new ArrayList<>();
        QueryWrapper wrapper = new QueryWrapper();
        wrapper.orderByDesc("create_time");
        List<ProcInfo> procInfoList = procInfoMapper.selectList(wrapper);
        if (CollectionUtil.isNotEmpty(procInfoList)) {
            Set<Integer> procIds = procInfoList.stream().map(item -> item.getId()).collect(Collectors.toSet());
            wrapper = new QueryWrapper();
            wrapper.in("proc_id", procIds);
            List<WorkMenu> workMenus = baseMapper.selectList(wrapper);
            Map<Integer, List<WorkMenu>> workMenusMap = new HashMap<>();
            List<WorkMenu> procWorkMenus;
            if (CollectionUtil.isNotEmpty(workMenus)) {
                for (WorkMenu workMenu : workMenus) {
                    procWorkMenus = workMenusMap.get(workMenu.getProcId());
                    procWorkMenus = procWorkMenus == null ? new ArrayList<>() : procWorkMenus;
                    procWorkMenus.add(workMenu);
                    workMenusMap.put(workMenu.getProcId(), procWorkMenus);
                }
            }
            ProcWithWorkMenusVO vo;
            for (ProcInfo procInfo : procInfoList) {
                vo = new ProcWithWorkMenusVO();
                vo.setId(procInfo.getId());
                vo.setName(procInfo.getName());
                if (CollectionUtil.isNotEmpty(workMenusMap.get(vo.getId()))) {
                    try {
                        vo.setChildren(changeTreePack(PojoUtils.listConvert(WorkMenuDto.class, workMenusMap.get(vo.getId())), false, false));
                    } catch (IllegalAccessException e) {
                        log.error("获取工作流菜单树异常，error:{}", e.getMessage());
                        throw new RRException("获取工作流菜单树异常！");
                    }
                }
                resultList.add(vo);
            }
        }
        return resultList;
    }



    private JobLinkAndJobDto packSaveOrUpdateData(JobNodeConfig jobNodeConfig, WorkFlowDep workFlowDep) {
        JobLinkAndJobDto jobLinkAndJobDto = new JobLinkAndJobDto();
        jobLinkAndJobDto.setProcId(workFlowDep.getProcId());
        jobLinkAndJobDto.setWorkFlowMenuId(workFlowDep.getWorkMenuId());
        WorkMenuDto workMenuDto = new WorkMenuDto();
        workMenuDto.setId(workFlowDep.getWorkMenuId());
        List<JobInfoDto> retJobInfoList = listJobList(workMenuDto);
        List<JobLinkDto> retJobLinkList = listJobLinkList(workMenuDto, 1);
        jobLinkAndJobDto.setEdges(retJobLinkList);
        jobLinkAndJobDto.setNodes(retJobInfoList);
        jobLinkAndJobDto.setJobNodeConfig(jobNodeConfig);
        return jobLinkAndJobDto;
    }

    private Map<String, Set<Long>> changeTreePackData(List<WorkMenuDto> listConvert, Long parentId) throws IllegalAccessException {
        String childrenFieldName = "children";

        // 获取根节点，即找出父节点为0或者为空对象
        List<WorkMenuDto> topList = new ArrayList<>();
        topList = listConvert.stream().filter(workMenuDto -> Objects.equals(parentId, workMenuDto.getParentId())).collect(Collectors.toList());

        // 将根节点从原始list移除，减少下次处理数据
        listConvert.removeAll(topList);

        // 递归封装树
        Map<String, Set<Long>> restMap = new HashMap<>();
        Set<Long> jobNodeSet = new HashSet<>();
        Set<Long> jobSet = new HashSet<>();
        Set<Long> menuSet = new HashSet<>();
        fillTree(topList, listConvert, childrenFieldName, jobNodeSet, jobSet, menuSet, true);
        restMap.put("jobNodeSet", jobNodeSet);
        restMap.put("jobSet", jobSet);
        restMap.put("menuSet", menuSet);
        return restMap;
    }

    private void fillTree(List<WorkMenuDto> topList, List<WorkMenuDto> listConvert, String childrenFieldName, Set<Long> jobNodeSet, Set<Long> jobSet, Set<Long> menuSet, boolean subWorkFlow) throws IllegalAccessException {
        for (int i = 0; i < topList.size(); i++) {
            WorkMenuDto parent = topList.get(i);
            menuSet.add(topList.get(i).getId());
            //如果是1：则是节点或者任务（没有下级结构）
            if (parent.getWorkflowType() == 1) {
                jobNodeSet.add(parent.getId());
                continue;
            }
            if (parent.getWorkflowType() == 2) {
                jobSet.add(parent.getId());
                continue;
            }
            List<WorkMenuDto> children = fillChildren(parent, listConvert, childrenFieldName, subWorkFlow,true);
            if (children.isEmpty()) {
                continue;
            }
            children = children.stream().filter(workMenuDto -> Objects.equals(parent.getWorkflowType(), 0)).collect(Collectors.toList());
            if (null == children || children.isEmpty()) {
                continue;
            }
            listConvert.removeAll(children);
            fillTree(children, listConvert, childrenFieldName,jobNodeSet,jobSet,menuSet, subWorkFlow);
        }
    }

    private List<WorkMenuDto> changeTreePack(List<WorkMenuDto> listConvert, boolean subWorkFlow,boolean subNode) throws IllegalAccessException {
        String parentFieldName = "parentId";
        String childrenFieldName = "children";

        // 获取根节点，即找出父节点为0或者为空对象
        List<WorkMenuDto> topList = new ArrayList<>();
        topList = listConvert.stream().filter(workMenuDto -> Objects.equals(0L, workMenuDto.getParentId())).collect(Collectors.toList());

        // 将根节点从原始list移除，减少下次处理数据
        listConvert.removeAll(topList);

        // 递归封装树
        fillTree(topList, listConvert, childrenFieldName, subWorkFlow,subNode);

        return topList;
    }

    private List<WorkMenuDto> changeTreePack(List<WorkMenuDto> listConvert, boolean subWorkFlow,boolean subNode,
                                             List<JobLink> jobLinks, List<WorkMenuJobRel> jobRels, List<JobNodeConf> jobNodeConfs, List<JobNodeInfo> jobNodeInfos, List<JobInfo> jobInfos,String transferCode) throws IllegalAccessException {
        String childrenFieldName = "children";

        // 获取根节点，即找出父节点为0或者为空对象
        List<WorkMenuDto> topList = new ArrayList<>();
        topList = listConvert.stream().filter(workMenuDto -> Objects.equals(0L, workMenuDto.getParentId())).collect(Collectors.toList());

        // 将根节点从原始list移除，减少下次处理数据
        listConvert.removeAll(topList);

        // 递归封装树
        fillTree(topList, listConvert, childrenFieldName, subWorkFlow, subNode, jobLinks, jobRels, jobNodeConfs, jobNodeInfos, jobInfos,transferCode);

        return topList;
    }

    private List<WorkMenuDto> changeTreePackSubWork(List<WorkMenuDto> listConvert, boolean subWorkFlow,boolean subNode) throws IllegalAccessException {
        String parentFieldName = "parentId";
        String childrenFieldName = "children";

        // 获取根节点，即找出父节点为0或者为空对象
        List<WorkMenuDto> topList = new ArrayList<>();
        topList = listConvert.stream().filter(workMenuDto -> Objects.equals(0, workMenuDto.getWorkType())).collect(Collectors.toList());

        // 将根节点从原始list移除，减少下次处理数据
        listConvert.removeAll(topList);

        // 递归封装树
        fillTree(topList, listConvert, childrenFieldName, subWorkFlow,subNode);

        return topList;
    }

    private void fillTree(List<WorkMenuDto> topList, List<WorkMenuDto> listConvert, String childrenFieldName, boolean subWorkFlow,boolean subNode) throws IllegalAccessException {
        for (int i = 0; i < topList.size(); i++) {
            WorkMenuDto parent = topList.get(i);
            //如果是1：则是节点或者任务（没有下级结构）
            if (parent.getWorkflowType() == 1 || parent.getWorkflowType() == 2) {
                continue;
            }
            List<WorkMenuDto> children = fillChildren(parent, listConvert, childrenFieldName, subWorkFlow, subNode);
            if (children.isEmpty()) {
                continue;
            }
            children = children.stream().filter(workMenuDto -> Objects.equals(parent.getWorkflowType(), 0)).collect(Collectors.toList());
            if (null == children || children.isEmpty()) {
                continue;
            }
            listConvert.removeAll(children);
            fillTree(children, listConvert, childrenFieldName, subWorkFlow, subNode);
        }
    }

    private void fillTree(List<WorkMenuDto> topList, List<WorkMenuDto> listConvert, String childrenFieldName, boolean subWorkFlow,boolean subNode,
                          List<JobLink> jobLinks, List<WorkMenuJobRel> jobRels, List<JobNodeConf> jobNodeConfs, List<JobNodeInfo> jobNodeInfos, List<JobInfo> jobInfos,String transferCode) throws IllegalAccessException {
        for (int i = 0; i < topList.size(); i++) {
            WorkMenuDto parent = topList.get(i);
            //如果是1：则是节点或者任务（没有下级结构）
            if (parent.getWorkflowType() == 1 || parent.getWorkflowType() == 2) {
                continue;
            }
            List<WorkMenuDto> children = fillChildren(parent, listConvert, childrenFieldName, subWorkFlow, subNode, jobLinks, jobRels, jobNodeConfs, jobNodeInfos, jobInfos,transferCode);
            if (children.isEmpty()) {
                continue;
            }
            children = children.stream().filter(workMenuDto -> Objects.equals(parent.getWorkflowType(), 0)).collect(Collectors.toList());
            if (null == children || children.isEmpty()) {
                continue;
            }
            listConvert.removeAll(children);
            fillTree(children, listConvert, childrenFieldName, subWorkFlow, subNode, jobLinks, jobRels,jobNodeConfs, jobNodeInfos, jobInfos,transferCode);
        }
    }

    private List<WorkMenuDto> fillChildren(WorkMenuDto parent, List<WorkMenuDto> listConvert, String childrenFieldName, boolean subWorkFlow,boolean subNode) throws IllegalAccessException {
        List<WorkMenuDto> childList = new ArrayList<>();
        Long id = parent.getId();

        for (int i = 0; i < listConvert.size(); i++) {
            WorkMenuDto t = listConvert.get(i);
            Long childParentId = t.getParentId();
            if (id.equals(childParentId)) {
                childList.add(t);
            }
        }
        //查找到当前这个节点对应的子节点或者是任务
        List<WorkMenuDto> newChildList = findNewJobOrWorkflow(parent, subWorkFlow,subNode);
        if (CollectionUtil.isNotEmpty(newChildList)) {
            childList.addAll(newChildList);
        }
        if (!childList.isEmpty()) {
            FieldUtils.writeDeclaredField(parent, childrenFieldName, childList, true);
        }
        return childList;
    }

    private List<WorkMenuDto> fillChildren(WorkMenuDto parent, List<WorkMenuDto> listConvert, String childrenFieldName, boolean subWorkFlow,boolean subNode,
                                           List<JobLink> jobLinks, List<WorkMenuJobRel> jobRels, List<JobNodeConf> jobNodeConfs,
                                           List<JobNodeInfo> jobNodeInfos, List<JobInfo> jobInfos,String transferCode) throws IllegalAccessException {
        List<WorkMenuDto> childList = new ArrayList<>();
        Long id = parent.getId();

        for (int i = 0; i < listConvert.size(); i++) {
            WorkMenuDto t = listConvert.get(i);
            Long childParentId = t.getParentId();
            if (id.equals(childParentId)) {
                childList.add(t);
            }
        }
        //查找到当前这个节点对应的子节点或者是任务
        List<WorkMenuDto> newChildList = findNewJobOrWorkflow(parent, subWorkFlow,subNode, jobLinks, jobRels, jobNodeConfs, jobNodeInfos, jobInfos,transferCode);
        if (CollectionUtil.isNotEmpty(newChildList)) {
            childList.addAll(newChildList);
            parent.setIsExist(true);
        }
        if (!childList.isEmpty()) {
            FieldUtils.writeDeclaredField(parent, childrenFieldName, childList, true);
        }
        return childList;
    }

    private List<WorkMenuDto> findNewJobOrWorkflow(WorkMenuDto parent, boolean subWorkFlow,boolean subNode, List<JobLink> jobLinks, List<WorkMenuJobRel> jobRels, List<JobNodeConf> jobNodeConfList,
                                                   List<JobNodeInfo> jobNodeInfos, List<JobInfo> jobInfos,String transferCode) {
        List<WorkMenuDto> childList = new ArrayList<>();
        //查询到对应的所有的节点或者对应的工作流拼装成子节点
        Integer paraentIsDisplay = getIsDisplays(transferCode,parent.getCreatePer());
        if (parent.getWorkType() == WorkFlowMenuType.WORKFLOW_FIELD.getCode()) {
            //查询到这个菜单对应的任务是否存在
            Optional<WorkMenuJobRel> first = jobRels.stream().filter(jr -> ObjectUtils.equals(jr.getWorkType().intValue(), 1) && ObjectUtils.equals(jr.getWorkMenuId(), parent.getId())).findFirst();
            if (first.isPresent()) {
                //依赖个数
                parent.setLayUsedCount(countLayNumber(first.get().getWorkFlowId(), jobLinks));
                parent.setJobId(first.get().getWorkFlowId().intValue());
            }
            //通过其下的节点是否有来进行判断
            List<WorkMenuJobRel> workMenuJobRels1 = jobRels.stream().filter(jr -> ObjectUtils.equals(jr.getWorkType().intValue(), 2) && ObjectUtils.equals(jr.getWorkMenuId(), parent.getId())).collect(Collectors.toList());
            if (CollectionUtil.isNotEmpty(workMenuJobRels1)) {
                parent.setIsSaveLinked(1);
            }

            Optional<JobNodeConf> nodeConfFirst = jobNodeConfList.stream().filter(nc ->
                    "schedule_type".equals(nc.getKey()) && nc.getJobNodeId().longValue() == parent.getId().longValue() && ObjectUtils.equals(nc.getJobType(), 1)
            ).findFirst();
            if (nodeConfFirst.isPresent()) {
                //是否调度配置
                parent.setScheduleType(Integer.valueOf(nodeConfFirst.get().getValue()));
            }

            if(subNode){
                //根据id查询对应的子节点数据信息
                List<WorkMenuJobRel> workMenuJobRels = jobRels.stream().filter(jr -> ObjectUtils.equals(jr.getWorkType().intValue(), 0) && ObjectUtils.equals(jr.getWorkMenuId(), parent.getId())).collect(Collectors.toList());
                if (CollectionUtil.isNotEmpty(workMenuJobRels)) {
                    for (WorkMenuJobRel workMenuJobRel : workMenuJobRels) {
                        Optional<JobNodeInfo> nodeInfoFirst = jobNodeInfos.stream().filter(ni -> ni.getId().longValue() == workMenuJobRel.getWorkFlowId().longValue()).findFirst();
                        if (nodeInfoFirst.isPresent()) {
                            JobNodeInfo jobNodeInfo = nodeInfoFirst.get();
                            //判断节点是否有对应的配置，如果没有，则不能进行配置
                            List<JobNodeConf> jobNodeConfs = jobNodeConfList.stream().filter(nc -> ObjectUtils.equals(nc.getJobType() , 0) && ObjectUtils.equals(nc.getJobNodeId() , jobNodeInfo.getId())).collect(Collectors.toList());
                            if(!Objects.equals(1,jobNodeInfo.getIsVirCopy())){
                                WorkMenuDto menuDto = new WorkMenuDto();
                                menuDto.setId(Long.valueOf(jobNodeInfo.getId()));
                                menuDto.setParentId(parent.getId());
                                menuDto.setName(jobNodeInfo.getName());
                                menuDto.setWorkflowType(1);
                                menuDto.setTypeId(jobNodeInfo.getTypeId());
                                menuDto.setJobId(jobNodeInfo.getJobId());
                                menuDto.setCreatePer(jobNodeInfo.getCreatePer());
                                //是否显示
                                Integer childIsDisplay = getIsDisplays(transferCode,jobNodeInfo.getCreatePer());
                                menuDto.setIsDisplay(childIsDisplay);
                                menuDto.setDescName(jobNodeInfo.getDesc());
                                menuDto.setIsVirCopy(jobNodeInfo.getIsVirCopy());
                                if(Objects.equals(JobType.SHELL.getCode(),jobNodeInfo.getTypeId())){
                                    menuDto.setHostUrl(jobNodeInfo.getHostUrl());
                                    menuDto.setPort(jobNodeInfo.getPort());
                                    menuDto.setUsername(jobNodeInfo.getUsername());
                                    menuDto.setPassword(jobNodeInfo.getPassword());
                                }
                                menuDto.setCreatePerName(userNameByUserId(jobNodeInfo.getCreatePer()));
                                if (CollectionUtil.isNotEmpty(jobNodeConfs)) {
                                    menuDto.setIsConfig(1);
                                }
                                childList.add(menuDto);
                            }
                        }
                    }
                }
            }
            //节点任务的显示权限
            if(CollectionUtil.isNotEmpty(childList)){
                long count = childList.stream().filter(o -> Objects.equals(o.getIsDisplay(), 1)).count();
                if(count == childList.stream().count()){
                    parent.setIsDisplay(1);
                }
            }else{
                parent.setIsDisplay(paraentIsDisplay);
            }
        }
        if (subWorkFlow) {
            if (parent.getWorkType() == WorkFlowMenuType.WORKFLOW_MGT_FIELD.getCode()) {
                //根据id查询对应的子节点数据信息
                List<WorkMenuJobRel> workMenuJobRels = jobRels.stream().filter(jr -> ObjectUtils.equals(jr.getWorkType(), 2) && ObjectUtils.equals(jr.getWorkMenuId(), parent.getId())).collect(Collectors.toList());
                if (CollectionUtil.isNotEmpty(workMenuJobRels)) {
                    for (WorkMenuJobRel workMenuJobRel : workMenuJobRels) {
                        Optional<JobInfo> first = jobInfos.stream().filter(ji -> ji.getId().longValue() == workMenuJobRel.getWorkFlowId().longValue()).findFirst();
                        if (first.isPresent()) {
                            JobInfo jobInfo = first.get();
                            WorkMenuDto menuDto = new WorkMenuDto();
                            menuDto.setId(Long.valueOf(jobInfo.getId()));
                            menuDto.setParentId(parent.getId());
                            menuDto.setName(jobInfo.getName());
                            menuDto.setWorkflowType(2);
                            menuDto.setCreatePer(jobInfo.getCreatePer());
//                            //是否显示
//                            Integer childIsDisplay = getIsDisplays(transferCode,jobInfo.getCreatePer());
//                            menuDto.setIsDisplay(childIsDisplay);
                            menuDto.setCreatePerName(userNameByUserId(jobInfo.getCreatePer()));
                            childList.add(menuDto);
                        }
                    }
                }
            }
        }
        return childList;
    }

    private Integer getIsDisplays(String transferCode, Integer createPer) {
        if(!Objects.equals("ALL",transferCode)){
            if(!Objects.equals(createPer,Integer.valueOf(transferCode))){
                return 1;
            }
        }
        return 0;
    }

    private List<WorkMenuDto> findNewJobOrWorkflow(WorkMenuDto parent, boolean subWorkFlow,boolean subNode) {
        List<WorkMenuDto> childList = new ArrayList<>();
        //查询到对应的所有的节点或者对应的工作流拼装成子节点
        if (parent.getWorkType() == WorkFlowMenuType.WORKFLOW_FIELD.getCode()) {
            //查询到这个菜单对应的任务是否存在

            WorkMenuJobRel workMenuJobRel1 = workMenuJobRelMapper.selectOne(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 1).eq("work_menu_id", parent.getId()));
            if (null != workMenuJobRel1) {
                //依赖个数
                parent.setLayUsedCount(countLayNumber(workMenuJobRel1.getWorkFlowId()));
                parent.setJobId(workMenuJobRel1.getWorkFlowId().intValue());
            }
            //通过其下的节点是否有来进行判断
            List<WorkMenuJobRel> workMenuJobRels1 = workMenuJobRelMapper.selectList(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 2).eq("work_menu_id", parent.getId()));
            if (CollectionUtil.isNotEmpty(workMenuJobRels1)) {
                parent.setIsSaveLinked(1);
            }
            JobNodeConf jobNodeConf = jobNodeConfMapper.selectOne(new QueryWrapper<JobNodeConf>().eq("`key`", "schedule_type")
                    .eq("job_node_id", parent.getId())
                    .eq("job_type", 1));
            if (null != jobNodeConf) {
                //是否调度配置
                parent.setScheduleType(Integer.valueOf(jobNodeConf.getValue()));
            }

            if(subNode){
                //根据id查询对应的子节点数据信息
                List<WorkMenuJobRel> workMenuJobRels = workMenuJobRelMapper.selectList(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 0).eq("work_menu_id", parent.getId()));
                if (CollectionUtil.isNotEmpty(workMenuJobRels)) {
                    for (WorkMenuJobRel workMenuJobRel : workMenuJobRels) {
                        JobNodeInfo jobNodeInfo = jobNodeInfoMapper.selectById(workMenuJobRel.getWorkFlowId());
                        if (null != jobNodeInfo) {
                            //判断节点是否有对应的配置，如果没有，则不能进行配置
                            List<JobNodeConf> jobNodeConfs = jobNodeConfMapper.selectList(new QueryWrapper<JobNodeConf>().eq("job_type", 0).eq("job_node_id", jobNodeInfo.getId()));
                            if(!Objects.equals(1,jobNodeInfo.getIsVirCopy())){
                                WorkMenuDto menuDto = new WorkMenuDto();
                                menuDto.setId(Long.valueOf(jobNodeInfo.getId()));
                                menuDto.setParentId(parent.getId());
                                menuDto.setName(jobNodeInfo.getName());
                                menuDto.setWorkflowType(1);
                                menuDto.setTypeId(jobNodeInfo.getTypeId());
                                menuDto.setJobId(jobNodeInfo.getJobId());
                                menuDto.setCreatePer(jobNodeInfo.getCreatePer());
                                menuDto.setCreatePerName(userNameByUserId(jobNodeInfo.getCreatePer()));
                                if (CollectionUtil.isNotEmpty(jobNodeConfs)) {
                                    menuDto.setIsConfig(1);
                                }
                                childList.add(menuDto);
                            }
                        }
                    }
                }
            }

        }
        if (subWorkFlow) {
            if (parent.getWorkType() == WorkFlowMenuType.WORKFLOW_MGT_FIELD.getCode()) {
                //根据id查询对应的子节点数据信息
                List<WorkMenuJobRel> workMenuJobRels = workMenuJobRelMapper.selectList(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 2).eq("work_menu_id", parent.getId()));
                if (CollectionUtil.isNotEmpty(workMenuJobRels)) {
                    for (WorkMenuJobRel workMenuJobRel : workMenuJobRels) {
                        JobInfo jobInfo = jobInfoMapper.selectById(workMenuJobRel.getWorkFlowId());
                        if (null != jobInfo) {
                            WorkMenuDto menuDto = new WorkMenuDto();
                            menuDto.setId(Long.valueOf(jobInfo.getId()));
                            menuDto.setParentId(parent.getId());
                            menuDto.setName(jobInfo.getName());
                            menuDto.setWorkflowType(2);
                            menuDto.setCreatePer(jobInfo.getCreatePer());
                            menuDto.setCreatePerName(userNameByUserId(jobInfo.getCreatePer()));
                            childList.add(menuDto);
                        }
                    }
                }
            }
        }
        return childList;
    }

    private String userNameByUserId(Integer createPer) {
        SysUserEntity sysUserEntity = sysUserMapper.selectById(createPer);
        if(null != sysUserEntity){
            return sysUserEntity.getUsername();
        }
        return null;
    }

    private Integer countLayNumber(Long jobId, List<JobLink> jobLinkList) {
        if(CollectionUtil.isNotEmpty(jobLinkList)){
            List<JobLink> jobLinks = jobLinkList.stream().filter(jl -> ObjectUtils.equals(jl.getLinkeType(), 1) &&  ObjectUtils.equals(jl.getDstNodeId(), jobId)).collect(Collectors.toList());
            List<JobLink> jobLinks1 = jobLinkList.stream().filter(jl -> ObjectUtils.equals(jl.getLinkeType(), 1) &&  ObjectUtils.equals(jl.getSrcNodeId(), jobId)).collect(Collectors.toList());
            if (CollectionUtil.isEmpty(jobLinks) && CollectionUtil.isEmpty(jobLinks1)) {
                return 0;
            } else {
                return 1;
            }
        }
        return 0;
    }

    private Integer countLayNumber(Long jobId) {
        List<JobLink> jobLinks = jobLinkMapper.selectList(new QueryWrapper<JobLink>().eq("linke_type", 1).eq("dst_node_id", jobId));
        List<JobLink> jobLinks1 = jobLinkMapper.selectList(new QueryWrapper<JobLink>().eq("linke_type", 1).eq("src_node_id", jobId));
        if (CollectionUtil.isEmpty(jobLinks) && CollectionUtil.isEmpty(jobLinks1)) {
            return 0;
        } else {
            return 1;
        }
    }

    @Override
    public WorkMenu addOldWorkFlowMenu(WorkMenu workMenu1, JobInfo jobInfo) {

        //获取数据开发子目录
        WorkMenu workMenu2 = this.getOne(new QueryWrapper<WorkMenu>()
                .eq("name", "数据开发")
                .eq("parent_id", workMenu1.getId())
                .eq("proc_id", jobInfo.getProcId()));


        WorkMenu workMenu = workMenuMapper.selectOne(new QueryWrapper<WorkMenu>()
                .eq("proc_id", jobInfo.getProcId())
                .eq("parent_id", workMenu2.getId())
                .eq("work_type", WorkFlowMenuType.WORKFLOW_FIELD.getCode())
                .eq("name", jobInfo.getName())
                .eq("root_id", workMenu1.getId())
                .eq("create_per", ShiroUtils.getUserId().intValue()));

        if (workMenu == null) {
            //在根目录中添加作业流文件夹
            WorkMenu workMenu3 = new WorkMenu();
            workMenu3.setProcId(jobInfo.getProcId());
            workMenu3.setParentId(workMenu2.getId());
            workMenu3.setWorkType(WorkFlowMenuType.WORKFLOW_FIELD.getCode());
            workMenu3.setName(jobInfo.getName());
            workMenu3.setCreateTime(LocalDateTime.now());
            workMenu3.setRootId(workMenu1.getId());
            workMenu3.setCreatePer(ShiroUtils.getUserId().intValue());
            workMenuMapper.insert(workMenu3);
            workMenu = workMenu3;
        }

        WorkMenuJobRel workMenuJobRel1 = workMenuJobRelMapper.selectOne(new QueryWrapper<WorkMenuJobRel>()
                .eq("proc_id", jobInfo.getProcId())
                .eq("work_type", 1)
                .eq("work_menu_id", workMenu.getId())
                .eq("work_flow_id", Long.valueOf(jobInfo.getId())));

        if (workMenuJobRel1 == null) {
            //为job和menu增加关联
            //任务和菜单的关联关系数据
            WorkMenuJobRel workMenuJobRel = new WorkMenuJobRel();
            workMenuJobRel.setProcId(jobInfo.getProcId());
            workMenuJobRel.setWorkType(1);
            workMenuJobRel.setWorkMenuId(workMenu.getId());
            workMenuJobRel.setWorkFlowId(Long.valueOf(jobInfo.getId()));
            workMenuJobRel.setCreateTime(LocalDateTime.now());
            workMenuJobRelMapper.insert(workMenuJobRel);
            workMenuJobRel1 = workMenuJobRel;
        }


        // 虚拟化一个开始节点和一个结束节点，并且保存到hdfs上
        saveStartNodeAndEndNode(jobInfo.getProcId().longValue(), jobInfo.getId(), WorkFlowType.WORKFLOW.getCode());

        return workMenu;
    }



    @Override
    public void updateRouteWorkFlowXml(Integer jobId) throws Exception {
        //重新生成（可能还得修改上个节点的shell文件）---暂时这个失败走哪里先不做，有待优化
        //查询到对应的连线和对应的路由配置情况
        JobInfo jobInfo = jobInfoMapper.selectById(jobId);
        if(null == jobInfo){
            throw new RRException("任务不存在!");
        }
        WorkMenuJobRel workMenuJobRel = workMenuJobRelMapper.selectOne(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 1).eq("work_flow_id", jobId));
        if(null == workMenuJobRel){
            throw new RRException("任务关联的菜单不存在!");
        }

        //连线
        List<JobLink> jobLinks = jobLinkMapper.selectList(new QueryWrapper<JobLink>().eq("linke_type", 0).eq("menu_id", workMenuJobRel.getWorkMenuId()));
        List<WorkMenuJobRel> workMenuJobRels = workMenuJobRelMapper.selectList(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 0).eq("work_menu_id", workMenuJobRel.getWorkMenuId()));
        Set<Integer> nodeIds = new HashSet<>();
        for (WorkMenuJobRel menuJobRel : workMenuJobRels) {
            nodeIds.add(menuJobRel.getWorkFlowId().intValue());
        }
        Set<JobNodeInfoDto> nodeDtos = new HashSet<>();
        if(CollectionUtil.isNotEmpty(nodeIds)){
            List<JobNodeInfoDto> jobNodeInfoDtos = jobNodeInfoMapper.selectListByIds(nodeIds);
            nodeDtos.addAll(jobNodeInfoDtos);
            nodeIds.clear();
            if(CollectionUtil.isNotEmpty(jobNodeInfoDtos)){
                nodeIds.addAll(jobNodeInfoDtos.stream().map(JobNodeInfoDto::getId).collect(Collectors.toSet()));
            }
        }

        //查询是否配置正常调度
        List<JobNodeConf> jobNodeConfs = jobNodeConfMapper.selectList(new QueryWrapper<JobNodeConf>().eq("job_type", 1).eq("job_node_id", workMenuJobRel.getWorkMenuId())
                .in("`key`", "expression", "schedule_type", "scheduleOrInterval","cronType","cronUrl"));
        JobNodeConfig jobNodeConfig = JobNodeConfig.listToJobNodeConfig(jobNodeConfs);

        generatedAutoWorkFlowXmlAndRoute(jobLinks,nodeIds,jobNodeConfig,Long.valueOf(jobInfo.getId()),jobInfo.getProcId(), nodeDtos);

    }

    private void upShellVir(Set<JobNodeInfoDto> nodeList, int type,HdfsUtil hdfsUtil,Integer procId,Integer jobId) {
        String dst = "workflow-app/" + WorkFlowType.WORKFLOW.getCode() + "/" + procId.toString() + "/" + jobId.toString();
        for (JobNodeInfoDto jobNodeInfoDto : nodeList) {
            String nodeIdDst = dst + "/" + jobNodeInfoDto.getWorkFlowNodeName() + ".sh";
            String shell = "";
            if(1 == type){
                //默认节点---默认节点跑错执行
                shell = "echo 'run defNode end'\nexit 10";
            }
            if(2 == type){
                //虚拟的传参节点
                shell = "virCount=$1\necho \"virCount:$virCount\"";
            }
            if(3 == type){
                //虚拟的空跑
                nodeIdDst = dst + "/" + VirtualTypeEnum.VIRTUAL_NULL.getName() + "_" + jobNodeInfoDto.getId() + ".sh";
                shell = "echo 'null run'\nvirCount=0\necho \"virCount:$virCount\"";
            }
            if(4 == type){
                //默认节点---虚拟的结束节点执行
                shell = "echo 'run defEndNode end'";
            }
            hdfsUtil.writeFile(shell.getBytes(),nodeIdDst);
        }

    }

    private List<List<JobNodeInfoDto>> listAllJobNodeDtoList(List<JobLink> newJobLinks, List<JSONObject> addLinks, List<JobNodeInfoDto> newJobNodeInfoDtos) {
        List<JSONObject> allListObj = new ArrayList<>();
        if(CollectionUtil.isNotEmpty(newJobLinks)){
            for (JobLink jobLink : newJobLinks) {
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("srcNodeId",jobLink.getSrcNodeId());
                jsonObject.put("dstNodeId",jobLink.getDstNodeId());
                allListObj.add(jsonObject);
            }
        }
        if(CollectionUtil.isNotEmpty(addLinks)){
            allListObj.addAll(addLinks);
        }
        Set<String> nodeIds = new HashSet<>();
        //虚拟的是以名字来取的
        Set<String> virNodeIds = new HashSet<>();
        for (JobNodeInfoDto jobNodeInfoDto : newJobNodeInfoDtos) {
            if(jobNodeInfoDto.getVirType() == null || 0 == jobNodeInfoDto.getVirType()){
                nodeIds.add(String.valueOf(jobNodeInfoDto.getId()));
            } else {
                nodeIds.add(jobNodeInfoDto.getWorkFlowNodeName());
                virNodeIds.add(jobNodeInfoDto.getWorkFlowNodeName());
            }
        }
        //找出所有的连线（包括虚拟的开始节点）
        List<List<String>> lists = listLinkedList1(allListObj, nodeIds);
        if (CollectionUtil.isEmpty(lists)) {
            return null;
        }

        //直接生成对应的表达式
        List<List<JobNodeInfoDto>> allLinkedDtoList = new ArrayList<>();
        JobNodeInfoDto startJobNodeInfoDto = new JobNodeInfoDto();
        startJobNodeInfoDto.setWorkFlowNodeName(JobType.ObjOf(0).getEnName() + "_0");
        for (List<String> list : lists) {
            List<JobNodeInfoDto> list1 = new ArrayList<>();
            list1.add(startJobNodeInfoDto);
            for (int i = 1; i < list.size(); i++) {
                int finalI = i;
                if(virNodeIds.contains(list.get(i))){
                    List<JobNodeInfoDto> collect = newJobNodeInfoDtos.stream().filter(jobNodeInfoDto -> Objects.equals(list.get(finalI), jobNodeInfoDto.getWorkFlowNodeName())).collect(Collectors.toList());
                    list1.add(collect.get(0));
                }else {
                    List<JobNodeInfoDto> collect = newJobNodeInfoDtos.stream().filter(jobNodeInfoDto -> Objects.equals(list.get(finalI), String.valueOf(jobNodeInfoDto.getId()))).collect(Collectors.toList());
                    list1.add(collect.get(0));
                }
            }
            allLinkedDtoList.add(list1);
        }
        return allLinkedDtoList;
    }

    private JSONObject pingDefNodeByVirl(Integer id) {
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("dstNodeId",VirtualTypeEnum.VIRTUAL_DEF.getName() + "_" + id);
        jsonObject.put("srcNodeId",VirtualTypeEnum.VIRTUAL_SUM.getName() + "_" + id);
        return jsonObject;
    }

    private JSONObject pingVirlNodeStart(Integer startId, Integer dstNodeId) {
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("dstNodeId",dstNodeId);
        jsonObject.put("srcNodeId", VirtualTypeEnum.VIRTUAL_SUM.getName() + "_" + startId);
        return jsonObject;
    }

    private JSONObject pingVirlNode(Integer id) {
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("dstNodeId", VirtualTypeEnum.VIRTUAL_SUM.getName() + "_" + id);
        jsonObject.put("srcNodeId",id);
        return jsonObject;
    }

    private JobNodeInfoDto pingVirlJobNodeInfoDto(Integer jobNodeInfoDtoId) {
        JobNodeInfoDto jobNodeInfoDto = new JobNodeInfoDto();
        jobNodeInfoDto.setWorkFlowNodeName(VirtualTypeEnum.VIRTUAL_SUM.getName() + "_" + jobNodeInfoDtoId);
        jobNodeInfoDto.setVirType(VirtualTypeEnum.VIRTUAL_SUM.getId());
        return jobNodeInfoDto;
    }

    private JobNodeInfoDto pingDefJobNodeInfoDto(Integer jobNodeId) {
        JobNodeInfoDto jobNodeInfoDto = new JobNodeInfoDto();
        jobNodeInfoDto.setWorkFlowNodeName(VirtualTypeEnum.VIRTUAL_DEF.getName() + "_" + jobNodeId);
        jobNodeInfoDto.setVirType(VirtualTypeEnum.VIRTUAL_DEF.getId());
        return jobNodeInfoDto;
    }

    private JSONObject pingDefNode(Integer jobNodeId) {
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("dstNodeId", VirtualTypeEnum.VIRTUAL_DEF.getName() + "_" + jobNodeId);
        jsonObject.put("srcNodeId",jobNodeId);
        return jsonObject;
    }


    /***
     * 挑选出对应的配置了路由的连线节点
     * @param jobNodeInfoDtos 所有的节点的配置信息
     * @param dstJobLinkList  需要判断的连线
     * @param unConfigList    未配置的数据集合
     * @return
     */
    private List<JobLink> filtCondition(List<JobNodeInfoDto> jobNodeInfoDtos, List<JobLink> dstJobLinkList, List<JobLink> unConfigList) {
        List<JobLink> restList = new ArrayList<>();
        for (JobLink jobLink : dstJobLinkList) {
            if(cheackSubNodeIsLayCondition(jobNodeInfoDtos.stream().filter(jobNodeInfoDto -> Objects.equals(jobNodeInfoDto.getId(),jobLink.getDstNodeId())).findAny().orElse(null))){
                restList.add(jobLink);
            } else {
                unConfigList.add(jobLink);
            }
        }
        return restList;
    }

    /***
     * 验证是否有配置对应的路由条件
     * @param jobNodeInfoDto
     * @return
     */
    private Boolean cheackSubNodeIsLayCondition(JobNodeInfoDto jobNodeInfoDto) {
        if(null == jobNodeInfoDto){
            return Boolean.FALSE;
        }
        List<JobNodeLayConfigDto> jobNodeLayConfigDtoList = jobNodeInfoDto.getJobNodeLayConfigDtoList();
        if(CollectionUtil.isNotEmpty(jobNodeLayConfigDtoList)){
            for (JobNodeLayConfigDto jobNodeLayConfigDto : jobNodeLayConfigDtoList) {
                if( Objects.equals(RouteTypeEnum.COUNT.getId(),jobNodeLayConfigDto.getConfigType()) && null != jobNodeLayConfigDto.getOutputTableId()){
                    return Boolean.TRUE;
                }
            }
        }
        return Boolean.FALSE;
    }


}
